Getting Started
Architecture
Service Platform
NServiceBus
Transports
Persistence
ServiceInsight
ServicePulse
ServiceControl
Monitoring
Shape the future
Samples

RabbitMQ Transport Scripting

Target Version: NServiceBus 7.x

Example code and scripts to facilitate deployment and operational actions against RabbitMQ.

These samples use the RabbitMQ.Client NuGet Package.

Since the RabbitMQ.Client is not CLS Compliant it is not possible to run this code within PowerShell.

Native Send

The native send helper methods

A send involves the following actions:

  • Create and serialize headers.
  • Write a message body directly to RabbitMQ.

In C#

Copy code|Copy usings|Edit
public static void SendMessage(string machineName, string queueName, string userName, string password, string messageBody, Dictionary<string, object> headers)
{
    using (var connection = OpenConnection(machineName, userName, password))
    using (var channel = connection.CreateModel())
    {
        var properties = channel.CreateBasicProperties();
        properties.MessageId = Guid.NewGuid().ToString();
        properties.Headers = headers;
        var body = Encoding.UTF8.GetBytes(messageBody);
        channel.BasicPublish(string.Empty, queueName, false, properties, body);
    }
}

static IConnection OpenConnection(string machine, string userName, string password)
{
    var connectionFactory = new ConnectionFactory
    {
        HostName = machine,
        Port = AmqpTcpEndpoint.UseDefaultPort,
        UserName = userName,
        Password = password,
    };

    return connectionFactory.CreateConnection();
}

Using the native send helper methods

Copy code|Copy usings|Edit
SendMessage(
    machineName: "MachineName",
    queueName: "QueueName",
    userName: "admin",
    password: "password",
    messageBody: "{\"Property\":\"PropertyValue\"}",
    headers: new Dictionary<string, object>
    {
        {
            "NServiceBus.EnclosedMessageTypes", "MyNamespace.MyMessage"
        }
    });

In this example, the value MyNamespace.MyMessage represents the .NET type of the message. See the headers documentation for more information on the EnclosedMessageTypes header.

Return message to source queue

The retry helper methods

This code shows an example of how to perform the following actions:

  • Read a message from the error queue.
  • Extract the failed queue from the headers.
  • Forward that message to the failed queue name so it can be retried.
Copy code|Copy usings|Edit
public static void ReturnMessageToSourceQueue(IConnection connection, string errorQueueName, string messageId)
{
    using (var channel = connection.CreateModel())
    {
        // Enable publisher confirms so that messages aren't removed from the error queue until the broker confirms it has accepted the new message
        channel.ConfirmSelect();

        BasicGetResult result;

        do
        {
            result = channel.BasicGet(errorQueueName, false);

            if (result == null || result.BasicProperties.MessageId != messageId)
            {
                continue;
            }
            ReadFailedQueueHeader(out var failedQueueName, result);

            channel.BasicPublish(string.Empty, failedQueueName, false, result.BasicProperties, result.Body);

            // Wait for confirmation that message is sent back to source queue
            channel.WaitForConfirmsOrDie();

            // Acknowledge and consume the incoming message
            channel.BasicAck(result.DeliveryTag, false);

            return;
        } while (result != null);

        throw new Exception($"Could not find message with id '{messageId}'");
    }
}

static void ReadFailedQueueHeader(out string queueName, BasicGetResult getResult)
{
    var headerBytes = (byte[])getResult.BasicProperties.Headers["NServiceBus.FailedQ"];
    var header = Encoding.UTF8.GetString(headerBytes);
    // In Versions 3.3.1 and below the machine name will be included after the @
    // In Versions 3.3.2 and above it will only be the queue name
    queueName = header.Split('@').First();
}

Using the retry helper methods

Copy code|Copy usings|Edit
using (var connection = connectionFactory.CreateConnection())
{
    ReturnMessageToSourceQueue(
        connection: connection,
        errorQueueName: "error",
        messageId: "6698f196-bd50-4f3c-8819-a49e0163d57b");
}

Create queues

Queue creation can be done for a specific endpoint or queues shared between multiple endpoints.

The create queue helper methods

Copy code|Copy usings|Edit
public static class QueueCreationUtils
{
    public static void CreateQueue(string uri, string queueName, bool durableMessages, bool createExchange)
    {
        var connectionFactory = new ConnectionFactory
        {
            Uri = new Uri(uri)
        };

        using (var connection = connectionFactory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            channel.QueueDeclare(
                queue: queueName,
                durable: durableMessages,
                exclusive: false,
                autoDelete: false,
                arguments: null);

            if (createExchange)
            {
                CreateExchange(channel, queueName, durableMessages);
            }
        }
    }

    static void CreateExchange(IModel channel, string queueName, bool durableMessages)
    {
        channel.ExchangeDeclare(queueName, ExchangeType.Fanout, durableMessages);
        channel.QueueBind(queueName, queueName, string.Empty);
    }
}

Creating queues for an endpoint

To create all queues for a given endpoint name.

Copy code|Copy usings|Edit
public static void CreateQueuesForEndpoint(string uri, string endpointName, bool durableMessages, bool createExchanges)
{
    // main queue
    QueueCreationUtils.CreateQueue(uri, endpointName, durableMessages, createExchanges);

    // callback queue
    QueueCreationUtils.CreateQueue(uri, $"{endpointName}.{Environment.MachineName}", durableMessages, createExchanges);

    // timeout queue - not used if using native timeouts
    QueueCreationUtils.CreateQueue(uri, $"{endpointName}.Timeouts", durableMessages, createExchanges);

    // timeout dispatcher queue - not used if using native timeouts
    QueueCreationUtils.CreateQueue(uri, $"{endpointName}.TimeoutsDispatcher", durableMessages, createExchanges);

    // retries queue
    // TODO: Only required in Versions 3 and below
    QueueCreationUtils.CreateQueue(uri, $"{endpointName}.Retries", durableMessages, createExchanges);

}

Using the create endpoint queues

Copy code|Copy usings|Edit
CreateQueuesForEndpoint(
    uri: "amqp://guest:guest@localhost:5672",
    endpointName: "myendpoint",
    durableMessages: true,
    createExchanges: true);

To create shared queues

Copy code|Copy usings|Edit
QueueCreationUtils.CreateQueue(
    uri: "amqp://guest:guest@localhost:5672",
    queueName: "error",
    durableMessages: true,
    createExchange: true);

QueueCreationUtils.CreateQueue(
    uri: "amqp://guest:guest@localhost:5672",
    queueName: "audit",
    durableMessages: true,
    createExchange: true);

Create HA policy

To configure HA policy, refer to the RabbitMQ HA documentation.

Delete queues

The delete helper queue methods

Copy code|Copy usings|Edit
public static class QueueDeletionUtils
{
    public static void DeleteQueue(string uri, string queueName)
    {
        var connectionFactory = new ConnectionFactory
        {
            Uri = new Uri(uri)
        };

        using (var connection = connectionFactory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            channel.QueueUnbind(queueName, queueName, string.Empty, null);
            channel.ExchangeDelete(queueName);
            channel.QueueDelete(queueName);
        }
    }
}

To delete all queues for a given endpoint

Copy code|Copy usings|Edit
public static void DeleteQueuesForEndpoint(string uri, string endpointName)
{
    // main queue
    QueueDeletionUtils.DeleteQueue(uri, endpointName);

    // callback queue
    QueueDeletionUtils.DeleteQueue(uri, $"{endpointName}.{Environment.MachineName}");

    // timeout queue - not used if using native timeouts
    QueueDeletionUtils.DeleteQueue(uri, $"{endpointName}.Timeouts");

    // timeout dispatcher queue - not used if using native timeouts
    QueueDeletionUtils.DeleteQueue(uri, $"{endpointName}.TimeoutsDispatcher");

    // retries queue
    // TODO: Only required in Versions 3 and below
    QueueDeletionUtils.DeleteQueue(uri, $"{endpointName}.Retries");
}
Copy code|Copy usings|Edit
DeleteQueuesForEndpoint(
    uri: "amqp://guest:guest@localhost:5672",
    endpointName: "myendpoint");

To delete shared queues

Copy code|Copy usings|Edit
QueueDeletionUtils.DeleteQueue(
    uri: "amqp://guest:guest@localhost:5672",
    queueName: "error");
QueueDeletionUtils.DeleteQueue(
    uri: "amqp://guest:guest@localhost:5672",
    queueName: "audit");

Related Articles

Feedback