MassTransit & RabbitMQ In Depth

// ❌ Synchronous chain — fragile
// Order API → Inventory API → Payment API → Email API
// One slow/failed hop blocks the whole request
// Retries hammer the caller; cascading failures spread

// ✅ Event-driven — resilient
// Order API publishes OrderPlaced
// Inventory, Payment, Email each consume independently
// Broker buffers spikes; consumers retry on their schedule
// Services scale and deploy independently

Use messaging when you need at-least-once delivery, fan-out to many subscribers, background processing, or sagas spanning multiple services. Skip it for simple CRUD where a single database transaction is enough.

// Good fits:
// - "Order placed" → notify 5 downstream systems
// - Heavy PDF generation off the HTTP thread
// - Microservices that must stay available when peers are down
// - Saga: reserve inventory → charge card → ship (multi-step)

// Skip messaging:
// - Single monolith with one SQL database
// - Read-only queries that need an immediate response

Producer  →  Exchange  →  (binding)  →  Queue  →  Consumer

// Exchange types:
// direct  — route by exact routing key (e.g. "orders.created")
// fanout  — broadcast to ALL bound queues (pub/sub)
// topic   — pattern match ("orders.*", "orders.#")
// headers — match on message headers (rare)

// Queue: durable buffer; messages wait until a consumer acks
// Binding: rule linking exchange → queue (key or pattern)

Messages survive broker restarts only when marked persistent and queues are durable. Consumers must ack after successful processing — otherwise RabbitMQ redelivers (at-least-once delivery).

// Publisher: mark message persistent
var props = channel.CreateBasicProperties();
props.Persistent = true;

// Consumer: manual ack (safer than auto-ack)
channel.BasicConsume(queue, autoAck: false, consumer);
// ... process message ...
channel.BasicAck(deliveryTag, multiple: false);

// Nack → requeue or dead-letter
channel.BasicNack(deliveryTag, multiple: false, requeue: false);

# docker-compose.yml
services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"    # AMQP
      - "15672:15672"  # Management UI → guest/guest
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

# docker compose up -d
# UI: http://localhost:15672

using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory { HostName = "localhost" };
await using var conn    = await factory.CreateConnectionAsync();
await using var channel = await conn.CreateChannelAsync();

await channel.QueueDeclareAsync(
    queue: "orders",
    durable: true, exclusive: false, autoDelete: false);

var body = Encoding.UTF8.GetBytes("""{"orderId":42}""");
await channel.BasicPublishAsync(
    exchange: "", routingKey: "orders", body: body);

Console.WriteLine("Published ✔");

var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (_, ea) =>
{
    var json = Encoding.UTF8.GetString(ea.Body.Span);
    try
    {
        await ProcessOrderAsync(json);
        await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
    }
    catch
    {
        // Don't requeue forever — send to DLQ in production
        await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false);
    }
};

await channel.BasicConsumeAsync("orders", autoAck: false, consumer);

// ✅ Consumer classes with DI (like controllers for messages)
// ✅ Automatic exchange/queue topology
// ✅ Retry, circuit breaker, rate limit middleware
// ✅ Request/response over messaging
// ✅ Saga state machines (long-running workflows)
// ✅ Outbox + EF Core integration
// ✅ In-memory transport for unit tests
// ✅ Same API for RabbitMQ, Azure Service Bus, Amazon SQS…

dotnet add package MassTransit
dotnet add package MassTransit.RabbitMQ

// Optional:
dotnet add package MassTransit.EntityFrameworkCore  // outbox
dotnet add package MassTransit.AspNetCore           // hosted service helpers

// appsettings.json
// "RabbitMq": { "Host": "localhost", "Username": "guest", "Password": "guest" }

builder.Services.AddMassTransit(x =>
{
    // Register all consumers in this assembly
    x.AddConsumers(typeof(Program).Assembly);

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration["RabbitMq:Host"], h =>
        {
            h.Username(builder.Configuration["RabbitMq:Username"]!);
            h.Password(builder.Configuration["RabbitMq:Password"]!);
        });

        // Auto-configure endpoints: OrderPlacedConsumer → queue for OrderPlaced
        cfg.ConfigureEndpoints(context);
    });
});

// IBus is your publish endpoint
// IPublishEndpoint / ISendEndpointProvider injected per-scope

Contracts are plain C# types in a shared project. Use record for immutable events. Namespace and name become the exchange routing identity.

// Contracts/OrderPlaced.cs — shared library
namespace MyApp.Contracts;

public record OrderPlaced(Guid OrderId, Guid CustomerId, decimal Total);
public record OrderCancelled(Guid OrderId, string Reason);

// Commands (point-to-point) vs Events (pub/sub):
// - Send → one consumer (ICommand)
// - Publish → all subscribers (IEvent)
// MassTransit treats both as messages; naming convention drives topology

app.MapPost("/orders", async (
    CreateOrderRequest req,
    IPublishEndpoint publish,
    AppDbContext db) =>
{
    var order = new Order { CustomerId = req.CustomerId, Total = req.Total };
    db.Orders.Add(order);
    await db.SaveChangesAsync();

    await publish.Publish(new OrderPlaced(order.Id, order.CustomerId, order.Total));

    return Results.Created($"/orders/{order.Id}", order);
});

public class OrderPlacedConsumer(
    IEmailService email,
    ILogger<OrderPlacedConsumer> log) : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        var msg = context.Message;
        log.LogInformation("Order {Id} placed for {Customer}", msg.OrderId, msg.CustomerId);

        await email.SendAsync(msg.CustomerId,
            $"Thanks! Order {msg.OrderId} total {msg.Total:C}");
    }
}

// MassTransit resolves OrderPlacedConsumer from DI
// ConfigureEndpoints wires: OrderPlaced → OrderPlacedConsumer

// Publish — fan-out event (0..N consumers)
await publish.Publish(new OrderPlaced(id, customerId, total));

// Send — point-to-point command (exactly one destination)
await sendEndpoint.Send(new ProcessPayment(orderId, amount),
    context => context.SetRoutingKey("payment-service"));

// Rule of thumb:
// Events = "something happened" → Publish
// Commands = "do this" → Send

x.UsingRabbitMq((context, cfg) =>
{
    cfg.Host("localhost");

    cfg.UseMessageRetry(r => r.Exponential(
        retryLimit: 5,
        minInterval: TimeSpan.FromSeconds(1),
        maxInterval: TimeSpan.FromMinutes(5),
        intervalDelta: TimeSpan.FromSeconds(2)));

    cfg.ConfigureEndpoints(context);
});

// Transient failures (HTTP 503, timeout) → retry
// Permanent failures (bad JSON) → skip to error queue fast

// Failed messages land in _error queues (visible in RabbitMQ UI)
// Consume faults for alerting:

public class OrderPlacedFaultConsumer : IConsumer<Fault<OrderPlaced>>
{
    public Task Consume(ConsumeContext<Fault<OrderPlaced>> context)
    {
        var ex = context.Message.Exceptions.FirstOrDefault();
        // Log, alert PagerDuty, store for manual replay
        return Task.CompletedTask;
    }
}

public record ValidateInventory(Guid ProductId, int Quantity);
public record InventoryResult(bool Available, int OnHand);

// Consumer responds inline:
public class ValidateInventoryConsumer : IConsumer<ValidateInventory>
{
    public async Task Consume(ConsumeContext<ValidateInventory> ctx)
    {
        var stock = await _repo.GetStockAsync(ctx.Message.ProductId);
        await ctx.RespondAsync(new InventoryResult(stock >= ctx.Message.Quantity, stock));
    }
}

// Caller:
var client = bus.CreateRequestClient<ValidateInventory>();
var response = await client.GetResponse<InventoryResult>(
    new ValidateInventory(productId, qty),
    timeout: RequestTimeout.After(s: 10));

var ok = response.Message.Available;

Request/response over a broker adds latency and ties up a response queue. If you need a synchronous answer in under 100ms, a direct HTTP call (or gRPC) is simpler. Use messaging request/response when the handler is slow, might be offline, or you want load leveling.

// Problem: SaveChanges succeeds, Publish crashes → DB updated, no event
// Solution: write message to Outbox table in SAME transaction

x.AddEntityFrameworkOutbox<AppDbContext>(o =>
{
    o.UseSqlServer();
    o.UseBusOutbox();   // background service publishes from outbox
});

// In consumer / endpoint — outbox enqueues instead of direct publish
// MassTransit delivers after commit — at-least-once, never lost

// See: github.com/stevsharp/OutboxPattern

public class PaymentConsumer(AppDbContext db) : IConsumer<ChargeCustomer>
{
    public async Task Consume(ConsumeContext<ChargeCustomer> ctx)
    {
        var id = ctx.MessageId ?? ctx.CorrelationId;  // unique per message

        if (await db.ProcessedMessages.AnyAsync(m => m.Id == id))
            return;   // already handled — safe no-op

        await _gateway.ChargeAsync(ctx.Message.OrderId, ctx.Message.Amount);

        db.ProcessedMessages.Add(new ProcessedMessage { Id = id });
        await db.SaveChangesAsync();
    }
}

// MassTransit saga tracks state across events:

public class OrderState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; } = "";
    public Guid OrderId { get; set; }
}

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public State Submitted { get; private set; } = null!;
    public State Completed  { get; private set; } = null!;
    public Event<OrderPlaced> OrderPlaced { get; private set; } = null!;
    public Event<PaymentCompleted> PaymentCompleted { get; private set; } = null!;

    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);
        Event(() => OrderPlaced, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => PaymentCompleted, x => x.CorrelateById(m => m.Message.OrderId));

        Initially(
            When(OrderPlaced)
                .Then(ctx => ctx.Saga.OrderId = ctx.Message.OrderId)
                .Publish(ctx => new ChargeCustomer(ctx.Saga.OrderId, ctx.Message.Total))
                .TransitionTo(Submitted));

        During(Submitted,
            When(PaymentCompleted).TransitionTo(Completed));
    }
}

// See: github.com/stevsharp/OutboxSagaDemo

builder.Services.AddHealthChecks()
    .AddRabbitMQ(
        rabbitConnectionString: "amqp://guest:guest@localhost:5672",
        name: "rabbitmq");

// MassTransit exposes bus health via AddMassTransit health checks
// Monitor: queue depth, consumer lag, error queue count
// RabbitMQ management plugin + Prometheus exporter

// No RabbitMQ needed in unit tests:
services.AddMassTransitTestHarness(cfg =>
{
    cfg.AddConsumer<OrderPlacedConsumer>();
});

var harness = provider.GetRequiredService<ITestHarness>();
await harness.Start();

await harness.Bus.Publish(new OrderPlaced(Guid.NewGuid(), Guid.NewGuid(), 99m));

Assert.True(await harness.Consumed.Any<OrderPlaced>());