MassTransit

The problem: repeated code around every message

In the RabbitMQ lesson, we saw how much code one simple message needs. In our online shop, we have dozens of message types. For each one, we must write these things:

1. Creating the Exchange, queue and Binding.
2. Turning the object into JSON and back.
3. Retrying when the email server does not answer for a moment.
4. Sending a broken message to the error queue.
5. Saving the message in the Outbox so it stays in sync with the database.
6. Creating a Scope for Dependency Injection for each message.

Each team writes these a little differently, and each one has a few small bugs. MassTransit has written all of these once, and correctly.

The main idea: your code is only business

1. A message is a simple record. It has no dependency on RabbitMQ.
2. A consumer is a class with one method. It implements the IConsumer interface.
3. The transport can be swapped. Today RabbitMQ, tomorrow Azure Service Bus. The consumer code does not change. For tests, there is also an in-memory transport.

Message and consumer

Keep the messages in a shared project (for example, Shop.Contracts). MassTransit knows the message type by its namespace and class name. So the namespace must be the same in the sender and the receiver.

namespace Shop.Contracts;

public sealed record OrderPlaced(Guid OrderId, string CustomerEmail, decimal Total);

The email service consumer:

using MassTransit;
using Shop.Contracts;

public sealed class SendReceiptConsumer(IEmailSender email) : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        var order = context.Message;
        // Idempotent: the sender skips orders that already got a receipt.
        await email.SendReceiptOnceAsync(order.OrderId, order.CustomerEmail, context.CancellationToken);
    }
}

If the Consume method finishes without an error, MassTransit Acks the message by itself. If it throws an error, the retry path starts.

Setup

using MassTransit;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();   // queue names like "send-receipt"
    x.AddConsumer<SendReceiptConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.UseMessageRetry(r =>
        {
            r.Intervals(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5)); // two retries
            r.Ignore<ArgumentException>();   // a bad message will not get better
        });

        cfg.ConfigureEndpoints(context);     // one queue per consumer, bound to its message types
    });
});

var app = builder.Build();
app.Run();

The ConfigureEndpoints method does a lot of work:

1. It creates one Exchange for each message type. Its name is built from the namespace and the class name.
2. It creates one queue for each consumer. The queue name is built from the consumer class name.
3. It connects the queue to the Exchange of its messages. So every new service that consumes OrderPlaced automatically gets its own copy.

Sending a message: Publish or Send?

• The Publish method is for events. “An order was placed”. The sender does not know who is listening. Zero, one or several consumers get it.
• The Send method is for commands. “Send this email”. It has one specific destination, and only one consumer does it.

app.MapPost("/orders", async (
    PlaceOrder cmd, ShopDbContext db, IPublishEndpoint publisher, CancellationToken ct) =>
{
    var order = Order.Place(cmd.CustomerEmail, cmd.Items);
    db.Orders.Add(order);

    await publisher.Publish(new OrderPlaced(order.Id, order.CustomerEmail, order.Total), ct);
    await db.SaveChangesAsync(ct); // order and message: one transaction

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

Retry and the error queue

1. The consumer threw an error. MassTransit gives the message to the same consumer a few more times, as configured.
2. A temporary error usually goes away. For example, the email server answers after a few seconds.
3. If all tries fail, the message goes to a queue with the error suffix. The error details are also saved in the message Headers.
4. The message is not lost. Someone checks the error queue. After the problem is fixed, they move the message back to the main queue.

Message and database together: Outbox

We saw the Dual Write problem in the Kafka lesson: the order was saved in the database, but the message was never sent. MassTransit has a ready Outbox with EF Core:

builder.Services.AddMassTransit(x =>
{
    x.AddEntityFrameworkOutbox<ShopDbContext>(o =>
    {
        o.UseSqlServer();
        o.UseBusOutbox();   // Publish writes to the outbox table, not to the broker
    });

    // ... consumers and UsingRabbitMq as before ...
});

public sealed class ShopDbContext(DbContextOptions<ShopDbContext> options) : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.AddInboxStateEntity();
        modelBuilder.AddOutboxMessageEntity();
        modelBuilder.AddOutboxStateEntity();
    }
}

Now the steps look like this:

1. The Publish method only puts the message in the Outbox table. Nothing has gone to RabbitMQ yet.
2. The SaveChangesAsync method saves the order and the message in one transaction. Either both, or neither.
3. A background service sends the messages in the table to RabbitMQ. If RabbitMQ is down, it sends them later.

The same package also has an Inbox for the consumer side. This Inbox keeps the IDs of messages it has seen and skips duplicate messages. You must turn it on separately for the consumer queue, with the same tables we created above.

Saga State Machine

For long jobs that involve several services, MassTransit has a state machine. The state of each order is saved in the database, and each message moves it one step forward. This topic is covered in the Saga lesson, with details and full code.

Common mistakes

When to use MassTransit?