RmToolkit.MessageBus 3.0.1

RmToolkit Message Bus & Saga — README

A compact guide for using the RmToolkit message bus and saga orchestration features.

Motivation

Modern modular monolith applications often need clear boundaries between functional modules (e.g., User, Order, Inventory) without fully distributing services.

The RmToolkit Message Bus provides:

• Message-based communication between modules
• Saga orchestration with compensation and fault handling
• Strong modular isolation with local reliability
• In-memory transport for development and testing

This enables modular architectures to evolve toward microservices later without major redesign, while preserving the simplicity and performance of a single-deployable monolith.

Overview

This repository contains a lightweight message-bus abstraction with support for:

• In-memory transport (configurable channel capacity)
• Saga (state machine) orchestration
• Filters for producers, publishers, senders, and consumers
• Built-in Result pattern support via RmToolkit.Result (included)
• Public contracts exposed via RmToolkit.MessageBus.Contracts

Packages

• RmToolkit.MessageBus
  Provides the message bus, DI extensions, transports, saga orchestration, filters, and runtime components.

• RmToolkit.MessageBus.Contracts
  Shared message contracts and interfaces (NuGet). Useful for external producers/consumers.

• RmToolkit.Result
  Included automatically for Result-pattern support.

Install contracts package if needed:

dotnet add package RmToolkit.MessageBus.Contracts

Registration / DI Configuration

Register the bus during application startup:

builder.Services.AddMessageBus(conf =>
{
    conf.UsingInMemory(opts =>
    {
        opts.ChannelCapacity = 1_000;
    });

    conf.ConsumerAssembliesToRegister = [typeof(Program).Assembly];

    conf.AddProducerFilter(typeof(CommonProduceFilter<>), b => b.ForMessage<IntegrationEvent1>());
    conf.AddPublishFilter(typeof(CommonPublishFilter<>));
    conf.AddSendFilter(typeof(CommonSendFilter<>));

    conf.AddConsumerFilter(typeof(ModuleAInboxFilter<>), 
        b => b.ForMessage<ModuleA.IntegrationEvent1>(order: 2));

    conf.AddSagaStateMachine<ThreeStepSagaOrchestrator, ThreeStepState>()
        .UseInMemory();
});

Notes

• ConsumerAssembliesToRegister controls consumer discovery.
• Saga persistence is pluggable (InMemory, MemoryCache, DistributedCache, EF Core).
• In-memory transport is ideal for development and tests.

Filters

Filters allow cross-cutting behavior around message flow.

public sealed class LoggingConsumeFilter<T> : IConsumeFilter<T>
{
    private readonly ILogger<LoggingConsumeFilter<T>> _logger;

    public LoggingConsumeFilter(ILogger<LoggingConsumeFilter<T>> logger)
    {
        _logger = logger;
    }

    public async Task Invoke(IConsumeContext<T> ctx, IPipe<T> next)
    {
        _logger.LogInformation("Handling {Message}", typeof(T).Name);
        await next.Invoke(ctx);
    }
}

Common uses

• Logging / telemetry
• Validation / authorization
• Tracing and correlation
• Retry orchestration

Consumers

Consumers implement IConsumer<TMessage>.

internal sealed class UserCreatedEventHandler 
    : IConsumer<UserCreated>
{
    private readonly ILogger<UserCreatedEventHandler> _logger;

    public UserCreatedEventHandler(ILogger<UserCreatedEventHandler> logger)
    {
        _logger = logger;
    }

    public async Task Handle(IConsumeContext<UserCreated> context)
    {
        await Task.Delay(500);
        _logger.LogInformation("UserCreated processed: {UserId}", context.Message.UserId);
    }
}

Compensation Consumer Example

public sealed class CompensateStepA 
    : IConsumer<Undo<DoStepA>>
{
    public async Task Handle(IConsumeContext<Undo<DoStepA>> context)
    {
        await context.PublishAsync(
            new Compensated<DoStepA> { CorrelationId = context.Message.CorrelationId }
        );
    }
}

Saga (Three-Step Example)

State

public sealed class ThreeStepState : ISaga
{
    public Guid CorrelationId { get; init; }
    public State CurrentState { get; set; } = State.None;
    public string? ErrorMessage { get; set; }
}

Messages

public record StartWorkflow : EventMessage;
public record DoStepA : CommandMessage;
public record DoStepB : CommandMessage;
public record DoStepC : CommandMessage;

public record StepACompleted : EventMessage;
public record StepBCompleted : EventMessage;
public record StepCCompleted : EventMessage;

Saga Definition (Excerpt)

Initially
    .When(StartWorkflowEvent)
    .Send(ctx => new DoStepA { CorrelationId = ctx.Message.CorrelationId })
    .TransitionTo(StepAInProgress);

Fault & Compensation Handling

• Consumers may publish Fault<T> messages on failure
• Sagas can react by:
  • Updating error state
  • Publishing Undo<T> compensation messages
  • Transitioning to a Failed state

Request / Response

Sending Commands

await bus.SendAsync(new CreateUserCommand(1, "ARMIN"));

Requesting a Response

var response = await bus.RequestAsync<GetUserQuery, UserDto>(
    new GetUserQuery(1),
    timeout: TimeSpan.FromMinutes(1),
    cancellationToken: cancellationToken
);

Responding in Consumer

public sealed class GetUserHandler : IConsumer<GetUserQuery>
{
    public async Task Handle(IConsumeContext<GetUserQuery> context)
    {
        await context.RespondAsync(new UserDto(1, "ARMIN"));
    }
}

Best Practices

• Prefer idempotent consumers
• Always set CorrelationId for sagas
• Keep saga state minimal and serializable
• Use compensation instead of distributed transactions

Troubleshooting

• Saga not correlating → Check CorrelationId and CorrelateById
• Consumers not discovered → Verify ConsumerAssembliesToRegister
• Filters not executing → Ensure filters are registered before bus startup

License

MIT License