AMQP.Net Lite Documentation

Azure Event Hubs supports an extended message format (0x80013700) that allows packing multiple messages into a single AMQP message. This is highly efficient for high-latency networks or small messages. The resulting AMQP message contains multiple Data sections, which the service extracts and delivers to receivers individually.

To implement this, you must set the Format to 0x80013700 and provide a DataList containing the serialized payloads in the BodySection.

public class MessageBatch : Message
{
    public const uint BatchFormat = 0x80013700;

    public static MessageBatch Create<T>(IEnumerable<T> objects)
    {
        DataList dataList = new DataList();
        foreach (var obj in objects)
        {
            ByteBuffer buffer = new ByteBuffer(1024, true);
            var section = new AmqpValue<T>(obj);
            AmqpSerializer.Serialize(buffer, section);
            dataList.Add(new Data() { Buffer = buffer });
        }

        return new MessageBatch() { Format = BatchFormat, BodySection = dataList };
    }
}

There are two primary patterns for receiving messages using a ReceiverLink:

1. Receive Loop (Manual)

Use a loop to call ReceiverLink.Receive(int). This call blocks until a message is available or the timeout elapses.

2. MessageCallback (Automatic)

Register a callback using ReceiverLink.Start(int, MessageCallback). This model eliminates the need for a manual receive loop in your application code.

Link Credit (Flow Control)

Link credit controls how many messages a remote peer can send.

• Automatic Mode (Default): The ReceiverLink manages credit. Credit is decremented when a message arrives and incremented when you call ReceiverLink.Accept(Message) or ReceiverLink.Reject(Message).
• Manual Mode: Call ReceiverLink.SetCredit(int, true) to set initial credit, or ReceiverLink.SetCredit(int, false) to take full control. If autoRestore is false, the library stops sending flow performatives, and you must manually renew credit via SetCredit.

The [AmqpProvides] attribute allows the decoder to resolve specific types based on a descriptor in the payload. This is useful for handling class inheritance or acting as a type registry.

Scenario 1: Class Inheritance If a base class and derived classes have the same EncodingType, you can annotate the base class with [AmqpProvides] for each known derived type. This allows AmqpSerializer.Deserialize<Base>(buffer) to return the correct concrete instance.

Scenario 2: Type Resolver (Registry) If you have multiple unrelated types that might appear in a payload, create a 'Resolver' class. Annotate the Resolver class with [AmqpProvides(typeof(T))] for every possible type. To decode, use the specialized method: AmqpSerializer.Deserialize<Resolver, TAs>(buffer) where TAs is the base type of all decoded objects.

Example:

[AmqpContract]
[AmqpProvides(typeof(Person))]
[AmqpProvides(typeof(Student))]
[AmqpProvides(typeof(Address))]
class Resolver
{
}

// Usage:
var obj = AmqpSerializer.Deserialize<Resolver, object>(buffer);

[AmqpContract]
class Student : Person
{
    [AmqpMember]
    public double GPA { get; set; }
}

[AmqpContract]
[AmqpProvides(typeof(Student))]
class Person
{
}

// Deserialization returns the concrete Student instance
var person = AmqpSerializer.Deserialize<Person>(buffer);

AMQP.Net Lite uses an asynchronous connection "pump" (a continuous loop that processes I/O) to handle sending and receiving. It is critical that your application does not block this pump, as doing so prevents the library from processing I/O, which typically leads to deadlocks, application hangs, or timeout errors.

There are two primary ways to accidentally block the pump:

1. Performing blocking operations in a callback: If you use a callback (e.g., in a Send method), do not perform long-running or blocking work (like Thread.Sleep) directly inside it. Instead, schedule the work asynchronously.
2. Blocking in an async continuation: If an await operation completes and its continuation runs synchronously on the same thread used by the pump, blocking that thread will halt all I/O. To prevent this, ensure continuations do not run on threads performing blocking work, or wrap async operations using a TaskCompletionSource with TaskCreationOptions.RunContinuationsAsynchronously.

Important: Avoid mixing the synchronous API with the asynchronous API on the same thread. For example, calling a synchronous .Send() immediately after an await .SendAsync() can cause the synchronous call to timeout because the pump cannot process the required acknowledgements.

// BAD: Blocking inside a callback
SenderLink sender = new SenderLink(session, "sender", "q1");
sender.Send(
    new Message("test"),
    (m, o, s) => Thread.Sleep(120000), // This blocks the pump
    sender);

// BAD: Mixing sync and async in a way that blocks the pump
SenderLink sender = new SenderLink(session, "sender", "q1");
await sender.SendAsync(new Message("m1"));
sender.Send(new Message("m2")); // This will likely timeout

ContainerHost is the simplest way to implement an AMQP listener, allowing you to specify multiple endpoints in a single host. The host listens on all transport endpoints for incoming connections and routes them based on the following logic:

1. Address Resolution: If an address resolver is set, the host translates the incoming attach address. This is useful for routing messages to different destinations or serving messages from multiple nodes.
2. Message/Request Routing: If a message or request processor is registered at the resolved address, the host creates a link endpoint and routes all received messages to that processor.
3. Link Routing: If no message/request processor matches but a ILinkProcessor is registered, the attach request is routed to that processor.
4. Rejection: If no match is found, the attach is rejected with an amqp:not-found error.

You can configure protocol behavior via the Listeners properties of the ContainerHost.