Queue, Topic, and Binding Management

TIP

Wolverine assumes that exchanges should be "fanout" unless explicitly configured otherwise

Use AutoProvision() to let Wolverine check for and create missing Rabbit MQ exchanges, queues, or bindings on demand, when it needs them. This enables Wolverine to manage the Rabbit MQ objects declared in the application configuration, but does not force every resource in the application to be set up during application startup.

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseRabbitMq(rabbit => { rabbit.HostName = "localhost"; })
            // I'm declaring an exchange, a queue, and the binding
            // key that we're referencing below.
            // This is NOT MANDATORY, but rather just allows Wolverine to
            // control the Rabbit MQ object lifecycle
            .DeclareExchange("exchange1", ex => { ex.BindQueue("queue1", "key1"); })

            // This will direct Wolverine to create any missing Rabbit MQ exchanges,
            // queues, or binding keys declared in the application at application
            // start up time
            .AutoProvision();

        opts.PublishAllMessages().ToRabbitExchange("exchange1");
    }).StartAsync();

^{snippet source | anchor}

At development time -- or occasionally in production systems -- you may want to have the messaging queues purged of any old messages at application startup time. Wolverine supports that with Rabbit MQ using the AutoPurgeOnStartup() declaration:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseRabbitMq()
            .AutoPurgeOnStartup();
    }).StartAsync();

^{snippet source | anchor}

Or you can be more selective and only have certain queues of volatile messages purged at startup as shown below:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseRabbitMq()
            .DeclareQueue("queue1")
            .DeclareQueue("queue2", q => q.PurgeOnStartup = true);
    }).StartAsync();

^{snippet source | anchor}

To force setup at application startup, use the JasperFx stateful resource model instead. AddResourceSetupOnStartup() registers an IHostedService that sets up every known JasperFx resource when the application starts. That includes Marten database objects, Wolverine durability database objects, and Wolverine message-broker objects. This works independently of AutoProvision().

Use AutoProvision() for on-demand Rabbit MQ provisioning, or use AddResourceSetupOnStartup() to provision all known JasperFx resources eagerly at startup. You can use both when you want eager startup setup plus on-demand provisioning for resources discovered later.

var builder = WebApplication.CreateBuilder(args);

builder.Host.ApplyJasperFxExtensions();

builder.Host.UseWolverine(opts =>
{
    // I'm setting this up to publish to the same process
    // just to see things work
    opts.PublishAllMessages()
        .ToRabbitExchange("issue_events", exchange => exchange.BindQueue("issue_events"))
        .UseDurableOutbox();

    opts.ListenToRabbitQueue("issue_events").UseDurableInbox();

    opts.UseRabbitMq(factory =>
    {
        // Just connecting with defaults, but showing
        // how you *could* customize the connection to Rabbit MQ
        factory.HostName = "localhost";
        factory.Port = 5672;
    })
        
    // Use for on-demand Rabbit MQ provisioning in addition to bootstrap time below
    .AutoProvision();
});

// Registers an IHostedService that sets up every known JasperFx resource
// at startup, including Marten/Wolverine database objects and Rabbit MQ objects
builder.Services.AddResourceSetupOnStartup();

// Just pumping out a bunch of messages so we can see
// statistics
builder.Services.AddHostedService<Worker>();

builder.Services.AddMarten(opts =>
{
    // I think you would most likely pull the connection string from
    // configuration like this:
    // var martenConnectionString = builder.Configuration.GetConnectionString("marten");
    // opts.Connection(martenConnectionString);

    opts.Connection(Servers.PostgresConnectionString);
    opts.DatabaseSchemaName = "issues";

    // Just letting Marten know there's a document type
    // so we can see the tables and functions created on startup
    opts.RegisterDocumentType<Issue>();

    // I'm putting the inbox/outbox tables into a separate "issue_service" schema
}).IntegrateWithWolverine(x => x.MessageStorageSchemaName = "issue_service");

var app = builder.Build();

app.MapGet("/", () => "Hello World!");

// Actually important to return the exit code here!
return await app.RunJasperFxCommands(args);

^{snippet source | anchor}

Note that this stateful resource model is also available at the command line as well for deploy time management.

Externally-Owned Queues and Exchanges 6.6

Sometimes a queue or exchange your application uses is owned and managed by a different system, and the identity your application connects with simply does not have the configure or delete permissions to create or remove it. In that case you want Wolverine to use the queue or exchange, but never try to declare it at startup (even with AutoProvision() turned on) and never delete it during a resource teardown. Mark the endpoint as ExternallyOwned() to get exactly that behavior:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseRabbitMq()
            // AutoProvision is on for the resources this app *does* own...
            .AutoProvision();

        // ...but this queue belongs to another team. Listen to it,
        // but never declare it at startup or delete it on teardown,
        // and don't try to set up or tear down its bindings.
        opts.ListenToRabbitQueue("shared-orders")
            .ExternallyOwned();

        // Same idea for an exchange we publish to but do not own
        opts.PublishMessage<OrderPlaced>()
            .ToRabbitExchange("shared-events")
            .ExternallyOwned();
    }).StartAsync();

When an endpoint is marked ExternallyOwned(), Wolverine will:

• Skip declaring the queue or exchange at startup, regardless of AutoProvision()
• Skip deleting it during a resources teardown (or the Oakton/JasperFx resource commands)
• Skip setting up or tearing down any bindings for that resource

This applies to queue listeners, queue subscribers, and exchange subscribers alike.

TIP

ExternallyOwned() is distinct from DeclarePassive. A DeclarePassive exchange still makes a passive declaration against the broker at startup to verify that the resource already exists (failing fast if it does not), whereas an externally-owned resource never touches the broker for declaration at all. As of Wolverine 6.6, a DeclarePassive exchange is also left alone during resource teardown — Wolverine will not delete a resource it only verified rather than created.

Exchange-to-Exchange Bindings

Wolverine supports RabbitMQ exchange-to-exchange bindings, which allow you to route messages between exchanges before they reach a queue. This is useful for building message routing topologies where a source exchange fans out to multiple destination exchanges, each with their own queue bindings.

You can declare exchange-to-exchange bindings using the fluent BindExchange().ToExchange() syntax:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseRabbitMq()
            .AutoProvision()

            // Bind source exchange to destination exchange with a routing key
            .BindExchange("source-exchange").ToExchange("destination-exchange", "routing.key")

            // The destination exchange still needs a queue binding for consumers
            .BindExchange("destination-exchange").ToQueue("my-queue", "routing.key");

        opts.PublishAllMessages().ToRabbitExchange("source-exchange");
        opts.ListenToRabbitQueue("my-queue");
    }).StartAsync();

When AutoProvision() is enabled, Wolverine declares the exchanges and creates the exchange-to-exchange bindings when they are needed. Both the source and destination exchanges are created if they do not already exist. Alternatively, add AddResourceSetupOnStartup() to force this setup during application startup.

Runtime Declaration

From a user request, there are some extension methods in the WolverineFx.RabbitMQ Nuget off of IWolverineRuntime that will enable you to first declare new exchanges, queues, and bindings at runtime, and also enable you to "unbind" a queue from an exchange. That syntax is shown below:

// _host is an IHost
var runtime = _host.Services.GetRequiredService<IWolverineRuntime>();

// Declare new Exchanges, Queues, and Bindings at runtime
await runtime.ModifyRabbitMqObjects(o =>
{
    var queue = o.DeclareQueue(queueName);
    var exchange = o.DeclareExchange(exchangeName);
    queue.BindExchange(exchange.ExchangeName, bindingKey);
});

// Unbind a queue from an exchange
await runtime.UnBindRabbitMqQueue(queueName, exchangeName, bindingKey);

^{snippet source | anchor}

Quorum Queues or Streams 3.10

Wolverine can utilize Rabbit MQ Quorum Queues or Rabbit MQ Streams, but "Classic" queues are the default. The only real difference as far as Wolverine is concerned is how the queues are declared to Rabbit MQ itself. Wolverine's internals are largely not impacted otherwise.

Here are your options for configuring one or many queues as opting into being a "Quorum Queue" or a "Stream":

var builder = Host.CreateApplicationBuilder();
builder.UseWolverine(opts =>
{
    opts
        .UseRabbitMq(builder.Configuration.GetConnectionString("rabbit")!)
        
        // You can configure the queue type for declaration with this
        // usage as well
        .DeclareQueue("stream", q => q.QueueType = QueueType.stream)

        // Use quorum queues by default as a policy
        .UseQuorumQueues()

        // Or instead use streams
        .UseStreamsAsQueues();

    opts.ListenToRabbitQueue("quorum1")
        // Override the queue type in declarations for a
        // single queue, and the explicit configuration will win
        // out over any policy or convention
        .QueueType(QueueType.quorum);
    
    
});

^{snippet source | anchor}

There are just a few things to know:

• Wolverine's internal reply or control queues will still be declared as "classic" so they can be non-durable
• Streams cannot be purged, and Wolverine ignores the AutoPurgeOnStartup() setting for streams

Inside of Wolverine Extensions

If you need to declare Rabbit MQ queues, exchanges, or bindings within a Wolverine extension, you can quickly access and make additions to the Rabbit MQ integration with your Wolverine application like so:

public class MyModuleExtension : IWolverineExtension
{
    public void Configure(WolverineOptions options)
    {
        options.ConfigureRabbitMq()
            // Make any Rabbit Mq configuration or declare
            // additional Rabbit Mq options through the normal
            // syntax
            .DeclareExchange("my-module")
            .DeclareQueue("my-queue");
    }
}

^{snippet source | anchor}

Identifier Prefixing for Shared Brokers

Because Rabbit MQ is a centralized broker model, you may need to share a single broker between multiple developers or development environments. To isolate the broker objects (queues, exchanges, bindings) created by each environment, you can use the PrefixIdentifiers() method to automatically prepend a prefix to every queue and exchange name created by Wolverine:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseRabbitMq()
            .AutoProvision()

            // Prefix all queue and exchange names with "dev-john-"
            .PrefixIdentifiers("dev-john");

        // A queue named "orders" becomes "dev-john-orders"
        opts.ListenToRabbitQueue("orders");
    }).StartAsync();

You can also use PrefixIdentifiersWithMachineName() as a convenience to use the current machine name as the prefix, which is often a good default for local development:

opts.UseRabbitMq()
    .AutoProvision()
    .PrefixIdentifiersWithMachineName();

The default delimiter between the prefix and the original name is - for Rabbit MQ (e.g., dev-john-orders).