Как работает MassTransit или другой .NET-клиент для брокеров?

namespace Shop.Contracts;

public sealed record OrderPlaced(Guid OrderId, Guid CustomerId, decimal Total, IReadOnlyList<OrderLine> Lines);
public sealed record OrderLine(string Sku, int Qty);
public sealed record ReserveStock(Guid OrderId, IReadOnlyList<OrderLine> Lines);    // команда
public sealed record StockStatusRequest(string Sku);                                // запрос
public sealed record StockStatusResponse(string Sku, int Available);

public sealed class ReserveStockConsumer(InventoryDbContext db, ILogger<ReserveStockConsumer> log) : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> ctx)
    {
        foreach (var line in ctx.Message.Lines)
        {
            var item = await db.Stock.SingleAsync(s => s.Sku == line.Sku, ctx.CancellationToken);
            item.Reserve(ctx.Message.OrderId, line.Qty);             // бросит OutOfStockException
        }
        await db.SaveChangesAsync(ctx.CancellationToken);
        await ctx.Publish(new StockReserved(ctx.Message.OrderId));   // событие дальше по цепочке
        log.LogInformation("Резерв для заказа {OrderId} выполнен", ctx.Message.OrderId);
    }
}

builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();                          // очередь "reserve-stock"
    x.AddConsumer<ReserveStockConsumer, ReserveStockConsumerDefinition>();
    x.AddConsumer<StockStatusConsumer>();

    x.AddEntityFrameworkOutbox<InventoryDbContext>(o =>
    {
        o.UsePostgres();
        o.UseBusOutbox();                                            // Publish/Send из обработчиков HTTP → в outbox
    });

    x.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.Host("rabbitmq", "/", h => { h.Username("app"); h.Password(builder.Configuration["Rabbit:Password"]!); });

        cfg.UseMessageRetry(r => r.Exponential(5, TimeSpan.FromMilliseconds(200), TimeSpan.FromSeconds(10), TimeSpan.FromMilliseconds(200)));
        cfg.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(30)));

        cfg.ConfigureEndpoints(ctx);                                // создать очереди и привязки для всех потребителей
    });
});

public sealed class ReserveStockConsumerDefinition : ConsumerDefinition<ReserveStockConsumer>
{
    protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpoint,
        IConsumerConfigurator<ReserveStockConsumer> consumer, IRegistrationContext context)
    {
        endpoint.PrefetchCount = 32;
        endpoint.ConcurrentMessageLimit = 8;                        // параллелизм обработки
        endpoint.UseMessageRetry(r => r.Ignore<OutOfStockException>().Interval(3, 500));   // бизнес-ошибку не ретраить
        endpoint.UseEntityFrameworkOutbox<InventoryDbContext>(context);   // inbox + outbox для потребителя
    }
}

// Команда
var ep = await sendEndpointProvider.GetSendEndpoint(new Uri("queue:generate-invoice"));
await ep.Send(new GenerateInvoice(orderId), ct);

// Запрос-ответ (с таймаутом)
var response = await stockClient.GetResponse<StockStatusResponse>(new StockStatusRequest("PHONE-1"), ct, timeout: RequestTimeout.After(s: 5));

public sealed class TenantFilter<T>(ITenantContext tenant) : IFilter<ConsumeContext<T>> where T : class
{
    public Task Send(ConsumeContext<T> ctx, IPipe<ConsumeContext<T>> next)
    {
        tenant.Set(ctx.Headers.Get<string>("tenant-id"));      // контекст из заголовков сообщения
        return next.Send(ctx);
    }
    public void Probe(ProbeContext ctx) => ctx.CreateFilterScope("tenant");
}
cfg.UseConsumeFilter(typeof(TenantFilter<>), ctx);

[Fact]
public async Task Publishes_StockReserved_after_successful_reservation()
{
    await using var provider = new ServiceCollection()
        .AddDbContext<InventoryDbContext>(o => o.UseNpgsql(fx.ConnectionString))
        .AddMassTransitTestHarness(x => x.AddConsumer<ReserveStockConsumer>())
        .BuildServiceProvider(true);

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

    await harness.Bus.Publish(new OrderPlaced(orderId, customerId, 100, [new("PHONE-1", 1)]));

    (await harness.Consumed.Any<OrderPlaced>()).Should().BeTrue();
    (await harness.Published.Any<StockReserved>(m => m.Context.Message.OrderId == orderId)).Should().BeTrue();
}