EventFlux.RabbitFlow

NuGet Package Information

Package	Downloads	License

EventFlux.RabbitFlow carries EventFlux events across services over RabbitMQ: publish an IEventRequest in one service and its IEventHandler<T> runs in another.

dotnet add package EventFlux.RabbitFlow --version 1.3.2

builder.Services.AddEventFluxRabbitFlow(
    typeof(Program).Assembly,
    new ConnectionFactory { Uri = new Uri("amqp://guest:guest@localhost:5672") },
    "orders_service",
    options => options.AutoSubscribe = true);

await eventBroker.PublishAsync(new OrderPlaced { OrderId = orderId }, cancellationToken);

Features

• Publish and subscribe to EventFlux events through RabbitMQ
• Handlers are plain EventFlux IEventHandler<T> classes, discovered by assembly scanning
• Subscriptions live for the whole lifetime of the host
• A failing handler never loses its message: it is requeued or moved to a dead-letter queue
• Optional limit on delivery attempts, with a delay between attempts
• Optional publisher confirms
• Channels are reused for publishing
• Graceful shutdown: messages being handled finish before the consumer stops
• Recovers from connection loss, broker restarts and closed channels without losing consumers or bindings
• Context propagation through message headers
• Cancellation support from publish to handler

Requirements

• .NET 8.0, 9.0 or 10.0
• EventFlux 2.1.1 or later
• A RabbitMQ broker reachable from your application

Installation

dotnet add package EventFlux.RabbitFlow --version 1.3.2

or in your .csproj:

<PackageReference Include="EventFlux.RabbitFlow" Version="1.3.2" />

Quick Start

1. Define an event

Create an event that implements IEventRequest:

public class ExampleEventRequest : IEventRequest
{
    public string Str { get; set; }
}

2. Create an event handler

Implement IEventHandler<T>. The cancellation token is the delivery's token, so it is cancelled when the consumer shuts down:

public class ExampleEventRequestHandler : IEventHandler<ExampleEventRequest>
{
    public Task Handle(ExampleEventRequest @event, CancellationToken cancellationToken)
    {
        Console.WriteLine("Consume data: " + @event.Str);
        return Task.CompletedTask;
    }
}

3. Configure Dependency Injection

AddEventFluxRabbitFlow registers EventFlux, scans the assembly for handlers and configures the broker. The third argument is the service name, which is also the default exchange:

builder.Services.AddEventFluxRabbitFlow(
    typeof(Program).Assembly,
    new ConnectionFactory { Uri = new Uri("amqp://guest:guest@localhost:5672") },
    "publisher_queue");

4. Subscribe to events

Either let the host subscribe every handler in the scanned assembly when it starts:

builder.Services.AddEventFluxRabbitFlow(assembly, connectionFactory, "publisher_queue",
    options => options.AutoSubscribe = true);

or subscribe explicitly after the app is built:

var app = builder.Build();

var eventBroker = app.Services.GetRequiredService<IEventBroker>();
await eventBroker.SubscribeAsync<ExampleEventRequest, ExampleEventRequestHandler>();

app.Services.UseEventFluxRabbitFlow(typeof(Program).Assembly) subscribes every handler of an assembly in one call.

IEventBroker is a singleton, so a subscription stays active until the host stops, no matter which scope created it. Subscribing the same handler twice is a no-op.

5. Publish an event

[HttpPost]
public async Task<IActionResult> Post(CancellationToken cancellationToken)
{
    var @event = new ExampleEventRequest { Str = "Hello World" };
    await _eventBroker.PublishAsync(@event, cancellationToken);
    return Ok();
}

6. Publish an event with context

[HttpPost]
public async Task<IActionResult> Post()
{
    _contextAccessor.Context = new EventFluxContext();
    _contextAccessor.Context.Items["UserId"] = "12345";
    _contextAccessor.Context.Items["CorrelationId"] = Guid.NewGuid().ToString();

    await _eventBroker.PublishAsync(new TestIntegrationEvent("aasdnasfnfavsfh"));

    return Ok();
}

7. Consume event with context

Inject IEventFluxContextAccessor into your event handler to read the context sent with the message:

public class ExampleEventRequestHandler : IEventHandler<ExampleEventRequest>
{
    private readonly IEventFluxContextAccessor _contextAccessor;

    public ExampleEventRequestHandler(IEventFluxContextAccessor contextAccessor)
    {
        _contextAccessor = contextAccessor;
    }

    public Task Handle(ExampleEventRequest @event, CancellationToken cancellationToken)
    {
        if (_contextAccessor.Context?.Items.TryGetValue("UserId", out var userId) == true)
        {
            Console.WriteLine($"UserId: {userId}");
        }

        return Task.CompletedTask;
    }
}

Failure handling

A message is acknowledged only after its handlers complete. When a handler throws, RabbitFlowOptions.OnFailure decides what happens:

OnFailure	Behaviour
Requeue (default)	The message is rejected and put back on its queue.
DeadLetter	The message is published to the dead-letter exchange and then acknowledged. If that publish fails, the message is requeued.

builder.Services.AddEventFluxRabbitFlow(assembly, connectionFactory, "orders_service", options =>
{
    options.OnFailure = RabbitFlowFailureMode.DeadLetter;
    options.DeadLetterExchange = "orders_failed";
});

In dead-letter mode every event queue {serviceName}_{EventName} gets a durable {serviceName}_{EventName}_dead_letter queue bound to the dead-letter exchange (default {serviceName}_dead_letter) with the event name as routing key. Dead-lettered messages keep their body and headers and add x-exception-type, x-exception-message, x-original-exchange and x-original-routing-key.

Limiting delivery attempts

By default a failing message is retried without limit, which keeps one poison message cycling through its queue. Set MaxDeliveryAttempts to move it to the dead-letter queue after a number of attempts (the first delivery counts as one), in either OnFailure mode:

builder.Services.AddEventFluxRabbitFlow(assembly, connectionFactory, "orders_service", options =>
{
    options.MaxDeliveryAttempts = 5;
    options.RedeliveryDelay = TimeSpan.FromSeconds(10);
});

• A failed message is sent back to the end of its own queue with an x-retry-count header, and the original is acknowledged. Other services bound to the same exchange do not receive it again.
• With RedeliveryDelay it first waits in a durable {serviceName}_{EventName}_retry queue and returns to the event queue when the delay has passed.
• After the last attempt it goes to the dead-letter queue with the headers above plus x-retry-count; x-original-exchange and x-original-routing-key keep the first route.
• If the retry or dead-letter publish fails, the message is requeued.
• Every consumer of the queue must run 1.2.0 or later before you enable it: an older consumer does not recognise retried messages and acknowledges them without handling.

A delivery cancelled by consumer shutdown is always requeued. A message for an event with no subscription is acknowledged and logged.

Configuration

Option	Default	Description
RetryCount	5	Connection and publish retry attempts, with exponential back-off.
OnFailure	Requeue	What happens to a message whose handler throws.
DeadLetterExchange	{serviceName}_dead_letter	Exchange used when OnFailure is DeadLetter.
PrefetchCount	0 (no limit)	Maximum unacknowledged messages per consumer channel.
AutoSubscribe	false	Subscribes every handler in the scanned assembly when the host starts, and stops consuming when it stops.
MaxDeliveryAttempts	null (no limit)	Attempts before a failing message is dead-lettered. See Limiting delivery attempts.
RedeliveryDelay	00:00:00	Wait between attempts when MaxDeliveryAttempts is set.
PublisherConfirms	false	PublishAsync completes only after the broker has confirmed the message. See Publisher confirms.
DeleteQueueOnUnsubscribe	false	When the last handler of an event is unsubscribed, delete its queue (and retry queue) instead of only unbinding it. Messages still in the queue are then lost.
ShutdownTimeout	00:00:30	How long disposing the broker waits for messages that are being handled.

Connection settings come from the IConnectionFactory passed to AddEventFluxRabbitFlow.

Shutdown

When the host stops, consumers are cancelled first, messages that arrived but have not started are requeued, and the messages already being handled are allowed to finish and be acknowledged before the channel closes. With AutoSubscribe this happens within the host's shutdown timeout; otherwise when the service provider disposes the broker, within ShutdownTimeout. A message still running when the time is up is delivered again after the restart, so handlers should stay idempotent.

Publisher confirms

With PublisherConfirms = true, PublishAsync waits until the broker has taken responsibility for the message:

• A message the broker rejects is retried like a connection failure, and the PublishException is thrown when RetryCount is exhausted.
• An event that no queue is bound to (no subscriber yet) is logged as a warning and does not fail.
• Messages carry an extra x-dotnet-pub-seq-no header added by the client. It is not passed to handlers, and older consumers ignore it.

Retry and dead-letter publishes made by the consumer are always confirmed when OnFailure = DeadLetter or MaxDeliveryAttempts is set: the original message is acknowledged only after the broker has accepted the copy, and if the copy cannot be routed (for example, the dead-letter queue was deleted) the original is requeued instead of lost.

Confirmations need the built-in connection. A custom IPersistenceConnection must implement CreateChannelAsync(CreateChannelOptions, CancellationToken); otherwise channels are created without confirmations and a warning is logged.

Connection recovery

• With ConnectionFactory.AutomaticRecoveryEnabled (the default), the client restores the connection, channels and consumers itself. RabbitFlow waits for it and does not open a second connection. Publishes made during the outage are retried until the connection is back or RetryCount is exhausted.
• With automatic recovery disabled, RabbitFlow opens a new connection, releases the old one, declares the exchanges, queues and bindings again (the service exchange is not durable, so a broker restart removes it) and restarts every consumer on it.
• If the broker closes only the consumer channel (for example after an invalid acknowledgement), consumers are restarted on a new channel, with or without automatic recovery.
• If the broker is still unreachable, restarting consumers is retried with back-off (up to 30 s between attempts) until it succeeds or the broker is disposed.
• A connection blocked by the broker (flow control) is logged, not reconnected.
• Connecting never blocks a thread: PublishAsync and SubscribeAsync wait for the connection asynchronously, and the CancellationToken passed to PublishAsync also cancels waiting for the broker. Custom IPersistenceConnection implementations can override TryConnectAsync; the default calls TryConnect.

Exchange Support

The service name passed to AddEventFluxRabbitFlow is the default exchange. Pass an exchange name to publish or subscribe on a different one.

await _eventBroker.PublishAsync(@event);
await _eventBroker.PublishAsync(@event, "eventbus_flow_2");

await _eventBroker.SubscribeAsync<ExampleEventRequest, ExampleEventRequestHandler>();
await _eventBroker.SubscribeAsync<ExampleEventRequest, ExampleEventRequestHandler>("eventbus_flow_2");

Message format

Messages are JSON bodies with the event type name as routing key, persisted on a direct exchange named after the service. Each subscribing service consumes from a durable queue named {serviceName}_{EventName}. The format is the same as in 1.0.x, so services on 1.0.x, 1.1.x, 1.2.x and 1.3.x can exchange messages. A retried message (see MaxDeliveryAttempts) is sent back to its own queue with the queue name as routing key; only 1.2.0 or later consumers understand it.

Upgrading from 1.3.0

1.3.1 only moves the EventFlux dependency to 2.1.1, a maintenance release with the same API and behaviour as 2.1.0. No code changes are needed.

Upgrading from 1.2.x

1.3.0 needs no code changes. Changes in behaviour:

• Unsubscribe no longer deletes the event queue when its last handler is removed: the queue is unbound and its consumer cancelled, and messages still waiting are kept for the next subscription. Set DeleteQueueOnUnsubscribe = true for the previous behaviour. Queues of events you no longer subscribe to stay on the broker until you delete them.
• On shutdown, messages that are being handled finish and are acknowledged before the channel closes, instead of being delivered again after the restart. Without AutoSubscribe this happens when the broker is disposed, bounded by ShutdownTimeout (30 seconds).
• Publishing reuses channels, so an application keeps up to 16 idle publish channels open instead of opening one per message.

Upgrading from 1.1.x

1.2.0 needs no code changes. Default behaviour is unchanged, apart from fixes:

• Connecting no longer blocks a thread, and the token passed to PublishAsync also cancels waiting for the broker.
• After a broker restart without automatic recovery, exchanges, queues and bindings are declared again before consumers restart. Previously messages published after the restart were dropped.
• A consumer channel closed by the broker is restarted, also with automatic recovery. Previously the application stopped consuming.
• In DeadLetter mode the dead-letter publish is confirmed by the broker before the original is acknowledged; if it cannot be routed, the message is requeued instead of lost.

New opt-in options: MaxDeliveryAttempts, RedeliveryDelay and PublisherConfirms. Enable MaxDeliveryAttempts only after every consumer of the queue runs 1.2.0.

Upgrading from 1.0.x

• Reference EventFlux 2.1.1 or later and target .NET 8.0, 9.0 or 10.0.
• Change handlers to Handle(TEvent @event, CancellationToken cancellationToken).
• Resolve IEventBroker from app.Services (or use AutoSubscribe) instead of building a second service provider with builder.Services.BuildServiceProvider().
• A failing handler no longer acknowledges its message. Make handlers idempotent, or use OnFailure = DeadLetter to stop a failing message from being retried.

Project layout

• EventBusRabbitMQ.cs — publish/subscribe orchestration and delivery handling
• EventBusSubscriptionsManager.cs — in-memory mapping between events and handlers
• PersistenceConnection.cs — RabbitMQ connection lifecycle
• RabbitFlowExtensions.cs — DI registration and subscription helpers
• RabbitFlowOptions.cs — configuration options