Fluens.Messaging 0.6.2

Fluens.Messaging

Inbox/outbox messaging pattern for modular applications.

Installation

dotnet add package Fluens.Messaging

Usage

Setup

// Host setup
fluensBuilder.AddMessaging();

// Module setup (inside IModuleSetup.RegisterServices)
// Consumers are auto-discovered from the module assembly
module.AddMessaging<OrdersDbContext>(msg =>
{
    msg.Publishes<OrderCreatedMessage>();
});

DbContext Configuration

Register the outbox interceptor and messaging entity configurations in your module DbContext:

// In AddDbContext registration
services.AddDbContext<OrdersDbContext>((sp, options) =>
{
    options.AddOutboxInterceptor(sp); // Enables atomic outbox writes
});

// In DbContext.OnModelCreating
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.HasDefaultSchema("orders");
    modelBuilder.ApplyConfigurationsFromAssembly(typeof(OrdersDbContext).Assembly);
    modelBuilder.ApplyMessagingConfigurations("orders"); // Registers OutboxMessage, InboxMessage, DeadLetter
}

Publishing

public record OrderCreatedMessage(Guid OrderId) : IMessage;

// Publish is synchronous — message is stored in outbox during SaveChangesAsync
publisher.Publish(new OrderCreatedMessage(order.ID));
await dbContext.SaveChangesAsync(); // Outbox interceptor writes message

Consuming

public class OrderCreatedConsumer : IMessageConsumer<OrderCreatedMessage>
{
    public async Task ConsumeAsync(
        OrderCreatedMessage message,
        MessageContext context,
        CancellationToken ct)
    {
        // Process the message
    }
}

Architecture: Publish (sync, in-memory) → Outbox (EF SaveChanges interceptor) → Transport (background worker) → Inbox (per-module processor) → Consumer. Failed messages are moved to a dead_letters table.

Configuration

MessagingOptions is bound from the "Fluens:Messaging" configuration section:

{
  "Fluens": {
    "Messaging": {
      "OutboxBatchSize": 100,
      "InboxBatchSize": 100,
      "InboxMaxRetries": 3,
      "InboxProcessorIntervalSeconds": 30,
      "TransportProcessorIntervalSeconds": 60
    }
  }
}

Graceful Shutdown

Both MessageTransportProcessor and InboxProcessor support cooperative cancellation:

• Batch loops check the cancellation token between iterations — unprocessed messages are picked up on next startup (at-least-once)
• OperationCanceledException during shutdown is not treated as a processing failure — no retry count increment, no dead letter moves
• Messages interrupted by shutdown retain their current retry count

License

This project is licensed under the MIT License.