CloudAMQP with .NET

We recommend the RabbitMQ .NET client, maintained by the RabbitMQ team. Version 7 has an async API, and recovers connections and consumers automatically.

dotnet new console --name AmqpExample
cd AmqpExample
dotnet add package RabbitMQ.Client

Publish and consume

Set CLOUDAMQP_URL to the AMQP URL from your instance’s details page in the CloudAMQP console. Without it, the examples connect to a local broker.

Save the connection helper as Amqp.cs:

using RabbitMQ.Client;

namespace AmqpExample;

public static class Amqp
{
    // Automatic recovery is on by default: the connection, channels and consumers are restored after a network failure
    public static Task<IConnection> ConnectAsync()
    {
        var url = Environment.GetEnvironmentVariable("CLOUDAMQP_URL") ?? "amqp://guest:guest@localhost";
        var factory = new ConnectionFactory { Uri = new Uri(url) };
        return factory.CreateConnectionAsync();
    }
}

Save the publisher as Publisher.cs:

using System.Text;
using RabbitMQ.Client;

namespace AmqpExample;

public static class Publisher
{
    // With confirmation tracking, BasicPublishAsync returns once the broker has confirmed the message.
    // This opens a channel per call; when publishing many messages, create the channel once and reuse it.
    public static async Task PublishAsync(IConnection connection, string queue, string message)
    {
        var options = new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true);
        await using var channel = await connection.CreateChannelAsync(options);
        await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false);

        var properties = new BasicProperties { Persistent = true };
        await channel.BasicPublishAsync("", queue, mandatory: true, properties, Encoding.UTF8.GetBytes(message));
        Console.WriteLine($" [x] Sent '{message}'");
    }
}

Save the consumer as Worker.cs:

using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace AmqpExample;

public static class Worker
{
    // The message is acked after the handler has processed it
    public static async Task ConsumeAsync(IChannel channel, string queue, Func<string, Task> handler)
    {
        await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false);
        await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false);

        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.ReceivedAsync += async (_, delivery) =>
        {
            await handler(Encoding.UTF8.GetString(delivery.Body.Span));
            await channel.BasicAckAsync(delivery.DeliveryTag, multiple: false);
        };
        await channel.BasicConsumeAsync(queue, autoAck: false, consumer);
    }
}

Replace the generated Program.cs with this entry point:

using AmqpExample;

await using var connection = await Amqp.ConnectAsync();

switch (args.FirstOrDefault())
{
    case "publish":
        await Publisher.PublishAsync(connection, "hello", "Hello CloudAMQP!");
        break;
    case "worker":
        await using (var channel = await connection.CreateChannelAsync())
        {
            await Worker.ConsumeAsync(channel, "hello", message =>
            {
                Console.WriteLine($" [x] Received '{message}'");
                return Task.CompletedTask;
            });
            Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");
            var exit = new TaskCompletionSource();
            Console.CancelKeyPress += (_, e) => { e.Cancel = true; exit.TrySetResult(); };
            await exit.Task;
        }
        break;
    default:
        Console.Error.WriteLine("Usage: AmqpExample publish|worker");
        return 1;
}
return 0;

Run dotnet run -- worker in one terminal and dotnet run -- publish in another.

A runnable example is in dotnet/rabbitmq-client in cloudamqp/examples.

Good to know

Connect with the URL. new ConnectionFactory { Uri = new Uri(url) }, where the URL is the one on your instance’s details page in the CloudAMQP console. With amqps:// the server certificate is verified.
Automatic recovery is on by default: after a network failure the client reconnects and restores channels, queues and consumers.
Version 7 is async-only. Code written for 6.x (CreateModel, EventingBasicConsumer) needs to move to the *Async methods; see the migration guide.
Ack after processing, and set a prefetch limit with BasicQosAsync, so a consumer doesn’t buffer more messages than it can handle.

← Rust

Crystal →