Azure Service Bus in .NET: publishing and consuming messages with a BackgroundService and DLQ

// ======================================================================
// Q: Show me publishing and consuming a message with Azure Service Bus in .NET.
//
// In larger systems I use a framework (MassTransit, NServiceBus, Wolverine):
// they provide ready-made retries, outbox, sagas and serialization. Watch the
// licenses: MassTransit v9 and NServiceBus are commercial.
// ======================================================================

// Program.cs: access via Managed Identity, no connection string (Q15)
builder.Services.AddAzureClients(b =>
{
    b.AddServiceBusClientWithNamespace(builder.Configuration["ServiceBus:Namespace"]!);
    b.UseCredential(new DefaultAzureCredential());
});
builder.Services.AddSingleton<OrderEventsPublisher>();
builder.Services.AddHostedService<OrderPlacedConsumer>();

// Publisher: ServiceBusClient and the sender are thread-safe, so singleton
public class OrderEventsPublisher(ServiceBusClient client)
{
    private readonly ServiceBusSender _sender = client.CreateSender("order-events");   // topic

    public Task PublishAsync(OrderPlaced e, CancellationToken ct) =>
        _sender.SendMessageAsync(new ServiceBusMessage(BinaryData.FromObjectAsJson(e))
        {
            MessageId = e.EventId.ToString(),        // for duplicate detection
            Subject   = nameof(OrderPlaced),         // message type, for routing / filters
            SessionId = e.OrderId.ToString()         // ordering (FIFO) per order, if the subscription has sessions
        }, ct);
}

// Consumer: BackgroundService, manual complete
public class OrderPlacedConsumer(ServiceBusClient client, IServiceScopeFactory scopes,
                                 ILogger<OrderPlacedConsumer> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await using var processor = client.CreateProcessor("order-events", "erp-sync",
            new ServiceBusProcessorOptions { AutoCompleteMessages = false, MaxConcurrentCalls = 4 });

        processor.ProcessMessageAsync += async args =>
        {
            var msg = args.Message;
            await using var scope = scopes.CreateAsyncScope();               // scoped DbContext (Q5)
            var handler = scope.ServiceProvider.GetRequiredService<IErpOrderSync>();
            try
            {
                var evt = msg.Body.ToObjectFromJson<OrderPlaced>()!;
                await handler.HandleAsync(evt, msg.MessageId, args.CancellationToken);
                await args.CompleteMessageAsync(msg, args.CancellationToken);       // success: removed from the queue
            }
            catch (Exception ex) when (ex is JsonException or ValidationException)
            {
                // Poison message: no matter how many retries, it will never succeed
                await args.DeadLetterMessageAsync(msg, "InvalidMessage", ex.Message, args.CancellationToken);
            }
            catch (Exception ex)
            {
                log.LogWarning(ex, "Processing {MessageId} failed, attempt {Count}", msg.MessageId, msg.DeliveryCount);
                await args.AbandonMessageAsync(msg, cancellationToken: args.CancellationToken);
                // redelivered; after MaxDeliveryCount it goes to the DLQ automatically
            }
        };
        processor.ProcessErrorAsync += e =>
        {
            log.LogError(e.Exception, "Service Bus error on {Entity}", e.EntityPath);
            return Task.CompletedTask;
        };

        await processor.StartProcessingAsync(ct);
        try { await Task.Delay(Timeout.Infinite, ct); }
        catch (OperationCanceledException) { }
        await processor.StopProcessingAsync();
    }
}