Benzene

RabbitMQ Setup

Benzene.RabbitMq is a self-hosted RabbitMQ consumer worker and outbound publish client, built on the RabbitMQ.Client v7 async API. It's Benzene's first vendor-neutral, self-hosted broker — every other broker integration is a cloud vendor's (SQS/SNS/Service Bus/Event Hubs/PubSub) plus Kafka — so it's the option for teams running on-prem, in Kubernetes, or across clouds who don't want to couple to a vendor's queue.

It shares the same [Message]/message-handler programming model as every other transport, and slots into the same self-hosted worker host as Benzene.Kafka.Core and Benzene.Azure.ServiceBus (see Unified Hosting Model) — Benzene owns the process, unlike the Lambda/Functions triggers.

Prerequisites

Install the NuGet packages

dotnet add package Benzene.RabbitMq --prerelease
dotnet add package Benzene.HostedService --prerelease
dotnet add package Benzene.SelfHost --prerelease

RabbitMQ.Client (v7) comes in transitively from Benzene.RabbitMq.

The worker assumes the queue exists

Benzene.RabbitMq does not declare exchanges, queues, or bindings — the worker consumes a queue you've already declared, exactly as the Kafka worker assumes its topics exist. Declare your topology out-of-band (the management UI, a definitions file, an IaC step, or a one-off channel.QueueDeclareAsync(...) at startup). Dead-letter topology is likewise yours to set up (see Ack policy below).

1. Define a message handler

The consumer routes each delivery to a handler by matching the message's topic against the handler's [Message("...")] value. The topic comes from the topic header, falling back to the AMQP routing key when that header isn't present (so a message published by a Benzene client routes by header, and a message from a non-Benzene producer routes by its natural routing key):

using Benzene.Abstractions.MessageHandlers;
using Benzene.Core.MessageHandlers;

[Message("orderCreated")]
public class OrderCreatedMessageHandler : IMessageHandler<OrderCreated>
{
    public Task HandleAsync(OrderCreated message)
    {
        Console.WriteLine($"Order {message.OrderId} created");
        return Task.CompletedTask;
    }
}

public class OrderCreated
{
    public string OrderId { get; set; }
}

This is an IMessageHandler<TRequest> (no response type) — the right shape for a fire-and-forget queue delivery. See Message Handlers for the request/response shape.

2. Define your StartUp

RabbitMQ consumption uses the same platform-neutral BenzeneStartUp as every other transport. The Benzene.RabbitMq.Extensions.UseRabbitMq extension targets IBenzeneWorkerStartup — the worker-specific builder that UseWorker(...) hands you — so you wire it up inside app.UseWorker(worker => worker.UseRabbitMq(...)):

using Benzene.Abstractions.Hosting;
using Benzene.Core.MessageHandlers;
using Benzene.Core.MessageHandlers.DI;
using Benzene.Microsoft.Dependencies;
using Benzene.RabbitMq;
using Benzene.SelfHost;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using RabbitMQ.Client;

public class StartUp : BenzeneStartUp
{
    public override IConfiguration GetConfiguration()
    {
        return new ConfigurationBuilder()
            .AddEnvironmentVariables()
            .Build();
    }

    public override void ConfigureServices(IServiceCollection services, IConfiguration configuration)
    {
        services.UsingBenzene(x => x
            .AddMessageHandlers(typeof(OrderCreatedMessageHandler).Assembly));
    }

    public override void Configure(IBenzeneApplicationBuilder app, IConfiguration configuration)
    {
        var config = new RabbitMqConfig
        {
            QueueName = "orders",
            PrefetchCount = 10,
            ConcurrentRequests = 5,
        };

        var connectionFactory = new RabbitMqConnectionFactory(new ConnectionFactory
        {
            HostName = "localhost",
            Port = 5672,
            UserName = "benzene",
            Password = "benzene",
        });

        app.UseWorker(worker =>
            worker.UseRabbitMq(config, connectionFactory, rabbit => rabbit.UseMessageHandlers()));
    }
}

3. Wire up Program.cs

Benzene.HostedService's UseBenzene<StartUp>() registers the RabbitMQ worker as an IHostedService, so it starts and stops with the host (a graceful stop cancels the consumer, drains in-flight handlers, then closes the channel and connection):

using Benzene.HostedService;

IHost host = Host.CreateDefaultBuilder(args)
    .UseBenzene<StartUp>()
    .Build();

await host.RunAsync();

Ack policy: what happens on failure

RabbitMqConfig.AckMode defaults to RabbitMqAckMode.Explicit — safe by default, unlike several other transports (see the Capability Matrix). RabbitMQ's first-class per-message ack makes this natural:

RabbitMQ's redelivered flag is a boolean, not a count, so the built-in bound is one retry; for a higher, precise limit configure a dead-letter exchange with a queue policy on the broker. RabbitMqAckMode.AutoAck is available for at-most-once, loss-tolerant workloads (the broker acks on dispatch, before the handler runs). Because redelivery can reprocess a message, handlers should be idempotent — see Idempotency.

Producing messages

To publish from another Benzene service (rather than a raw RabbitMQ.Client channel), use RabbitMqBenzeneMessageClient — an IBenzeneMessageClient, so your business logic depends only on the transport-agnostic client/IBenzeneMessageSender:

using Benzene.Clients;
using Benzene.RabbitMq.RabbitMqSendMessage;
using Microsoft.Extensions.Logging.Abstractions;
using RabbitMQ.Client;

await using var connection = await connectionFactory.CreateConnectionAsync(CancellationToken.None);
await using var channel = await connection.CreateChannelAsync();

var client = new RabbitMqBenzeneMessageClient(channel,
    NullLogger<RabbitMqBenzeneMessageClient>.Instance, serviceResolver, exchange: "");

await client.SendMessageAsync<OrderCreated, Benzene.Abstractions.Results.Void>(
    "orderCreated", new OrderCreated { OrderId = "123" });

The request Topic becomes the AMQP routing key and is also carried as a topic header, so a Benzene consumer routes by header (portable) with the routing key as the idiomatic fallback. The Benzene header dictionary is forwarded onto BasicProperties.Headers, so correlation-ID / W3C trace-context / payload-version headers reach the wire. With the default exchange (exchange: "") the routing key must equal the target queue name, so publishing topic "orders" lands in queue "orders"; for topic/direct/fanout exchanges, pass the exchange name and bind your queues to it out-of-band.

Publishing is fire-and-forget by default — a completed publish maps to Accepted, a thrown publish to ServiceUnavailable. (Publisher confirms for at-least-once publish are a planned opt-in.)

For the OutboundRoutingBuilder path (so call sites use IBenzeneMessageSender.SendAsync), the .UseRabbitMq<T>(exchange, ...) / .UseRabbitMqClient(channel) pipeline extensions are the conversion entry points, mirroring Kafka's .UseKafka<T>(...) — see Clients.

Testing

Benzene.RabbitMq's own tests show both levels:

Troubleshooting

See Also