Listener Pattern — ServiceBusSessionProcessor + DLQ Processor

Core Principles

• Listener is an IHostedService with constructor-initialized processors
• ServiceBusSessionProcessor for ordered per-aggregate processing
• Paired ServiceBusProcessor for dead letter queue (DLQ) reprocessing
• Subject-based routing via switch expression to typed HandleAsync<T> methods
• Typed deserialization: JsonConvert.DeserializeObject<Event<T>>(json)
• MediatR dispatch via scoped IServiceProvider for each message
• Return true from handler = CompleteMessageAsync, false = AbandonMessageAsync
• Serilog LogContext.Push with PropertyEnricher for structured logging

Key Patterns

Full Listener Implementation

namespace {Company}.{Domain}.Query.Infra.Services.ServiceBus.Listeners;

public class OrdersListener : IHostedService
{
    private readonly ILogger<OrdersListener> _logger;
    private readonly DomainServiceBus _serviceBusClient;
    private readonly IServiceProvider _serviceProvider;
    private readonly ServiceBusSessionProcessor _processor;
    private readonly ServiceBusProcessor _deadLetterProcessor;

    public OrdersListener(
        IServiceProvider serviceProvider,
        ILogger<OrdersListener> logger,
        IOptions<ServiceBusOptions> serviceBusOptions,
        DomainServiceBus serviceBusClient)
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
        _serviceBusClient = serviceBusClient;

        // Session processor configuration
        var options = new ServiceBusSessionProcessorOptions
        {
            AutoCompleteMessages = false,
            PrefetchCount = 1,
            MaxConcurrentCallsPerSession = 1,
            MaxConcurrentSessions = 1000,
            ReceiveMode = ServiceBusReceiveMode.PeekLock
        };

        _processor = _serviceBusClient.Client.CreateSessionProcessor(
            serviceBusOptions.Value.Topic,
            serviceBusOptions.Value.Subscription,
            options);

        _processor.ProcessMessageAsync += Processor_ProcessMessageAsync;
        _processor.ProcessErrorAsync += Processor_ProcessErrorAsync;

        // Dead letter queue processor
        _deadLetterProcessor = _serviceBusClient.Client.CreateProcessor(
            serviceBusOptions.Value.Topic,
            serviceBusOptions.Value.Subscription,
            new ServiceBusProcessorOptions()
            {
                PrefetchCount = 1,
                AutoCompleteMessages = false,
                MaxConcurrentCalls = 1,
                SubQueue = SubQueue.DeadLetter
            });

        _deadLetterProcessor.ProcessMessageAsync +=
            DeadLetterProcessor_ProcessMessageAsync;
        _deadLetterProcessor.ProcessErrorAsync +=
            DeadLetterProcessor_ProcessErrorAsync;
    }

    // --- Main message processing ---

    private async Task Processor_ProcessMessageAsync(
        ProcessSessionMessageEventArgs arg)
    {
        Task<bool> isHandledTask =
            HandleSubject(arg.Message.Subject, arg.Message);

        var isHandled = await isHandledTask;

        if (isHandled)
        {
            await arg.CompleteMessageAsync(arg.Message);
        }
        else
        {
            await arg.AbandonMessageAsync(arg.Message);
        }
    }

    private Task Processor_ProcessErrorAsync(ProcessErrorEventArgs arg)
    {
        _logger.LogCritical(arg.Exception,
            "OrdersListener => _processor => " +
            "Processor_ProcessErrorAsync " +
            "Message handler encountered an exception," +
            " Error Source:{ErrorSource}," +
            " Entity Path:{EntityPath}",
            arg.ErrorSource.ToString(),
            arg.EntityPath);

        return Task.CompletedTask;
    }

    // --- Dead letter processing (same routing) ---

    private async Task DeadLetterProcessor_ProcessMessageAsync(
        ProcessMessageEventArgs arg)
    {
        Task<bool> isHandledTask =
            HandleSubject(arg.Message.Subject, arg.Message);

        var isHandled = await isHandledTask;

        if (isHandled)
            await arg.CompleteMessageAsync(arg.Message);
        else
            await arg.AbandonMessageAsync(arg.Message);
    }

    private Task DeadLetterProcessor_ProcessErrorAsync(
        ProcessErrorEventArgs arg)
    {
        _logger.LogCritical(arg.Exception,
            "DeadLetter Message handler encountered an exception," +
            " Error Source:{ErrorSource}," +
            " Entity Path:{EntityPath}",
            arg.ErrorSource.ToString(),
            arg.EntityPath);

        return Task.CompletedTask;
    }

    // --- Subject-based routing with LogContext ---

    private Task<bool> HandleSubject(
        string subject, ServiceBusReceivedMessage message)
    {
        var eventType = new PropertyEnricher(
            name: "EventType", message.Subject);
        var sessionId = new PropertyEnricher(
            name: "SessionId", message.SessionId);
        var messageId = new PropertyEnricher(
            name: "MessageId", message.MessageId);

        using (LogContext.Push(eventType, sessionId, messageId))
        {
            return subject switch
            {
                EventType.OrderCreated =>
                    HandleAsync<OrderCreatedData>(message),
                EventType.OrderUpdated =>
                    HandleAsync<OrderUpdatedData>(message),
                EventType.OrderStatusChanged =>
                    HandleAsync<OrderStatusChangedData>(message),
                EventType.OrderItemAdded =>
                    HandleAsync<OrderItemAddedData>(message),
                // Skip known but unhandled event types
                EventType.OrderViewed => Task.FromResult(true),
                // Unknown events — return false
                _ => Task.FromResult(false)
            };
        }
    }

    // --- Typed deserialization + MediatR dispatch ---

    private async Task<bool> HandleAsync<T>(
        ServiceBusReceivedMessage message)
    {
        var json = Encoding.UTF8.GetString(message.Body);

        var body = JsonConvert.DeserializeObject<Event<T>>(json)
            ?? throw new InvalidOperationException(
                "Message body was serialized to null");

        using var scope = _serviceProvider.CreateScope();

        var mediator = scope.ServiceProvider
            .GetRequiredService<IMediator>();

        return await SendAsync(mediator, body);
    }

    private async Task<bool> SendAsync<T>(
        IMediator mediator, Event<T> @event)
    {
        var result = await mediator.Send(@event);

        if (!result)
            _logger.LogWarning(
                "Event Not Handling With Aggregate : {id}, " +
                "Type : {type} and sequence : {sequence}",
                @event.AggregateId,
                @event.Type,
                @event.Sequence);

        return result;
    }

    // --- Lifecycle ---

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _processor.StartProcessingAsync(cancellationToken);
        _deadLetterProcessor.StartProcessingAsync(cancellationToken);
        return Task.CompletedTask;
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        _processor.CloseAsync(cancellationToken);
        _deadLetterProcessor.CloseAsync(cancellationToken);
        return Task.CompletedTask;
    }
}

ServiceBus Client Wrapper

namespace {Company}.{Domain}.Query.Infra.Services.ServiceBus.Clients;

public class DomainServiceBus(ServiceBusOptions serviceBusOptions)
{
    public ServiceBusClient Client { get; set; } =
        new ServiceBusClient(serviceBusOptions.ConnectionString);
}

ServiceBus Options

namespace {Company}.{Domain}.Query.Infra.Services.ServiceBus;

public class ServiceBusOptions
{
    public const string ServiceBus = "ServiceBus";

    [Required]
    public required string ConnectionString { get; init; }

    [Required]
    public required string Topic { get; init; }

    [Required]
    public required string Subscription { get; init; }
}

EventType Constants

namespace {Company}.{Domain}.Query.Infra.Constants;

public class EventType
{
    public const string OrderCreated = "OrderCreated";
    public const string OrderUpdated = "OrderUpdated";
    public const string OrderStatusChanged = "OrderStatusChanged";
    public const string OrderItemAdded = "OrderItemAdded";
    public const string OrderViewed = "OrderViewed";
}

Registration

// In ServiceBusRegistrationExtensions
public static void UseServiceBus(
    this IServiceCollection services,
    IConfiguration configuration)
{
    services.AddOptions<ServiceBusOptions>()
        .Bind(configuration.GetSection(ServiceBusOptions.ServiceBus))
        .ValidateDataAnnotations()
        .ValidateOnStart();

    services.AddSingleton(provider =>
    {
        var options = provider
            .GetRequiredService<IOptions<ServiceBusOptions>>();
        return new DomainServiceBus(options.Value);
    });

    services.AddHostedService<OrdersListener>();
}

Session Processor Configuration

Setting	Value	Why
AutoCompleteMessages	false	Explicit Complete/Abandon based on handler result
PrefetchCount	1	Process one message at a time per session
MaxConcurrentCallsPerSession	1	Strict ordering within each aggregate session
MaxConcurrentSessions	1000	High parallelism across different aggregates
ReceiveMode	PeekLock	Message stays locked until Complete/Abandon

DLQ Processor Configuration

Setting	Value	Why
PrefetchCount	1	Conservative for dead letters
AutoCompleteMessages	false	Explicit Complete/Abandon
MaxConcurrentCalls	1	Sequential DLQ processing
SubQueue	SubQueue.DeadLetter	Read from dead letter sub-queue

Anti-Patterns

Anti-Pattern	Correct Approach
AutoCompleteMessages = true	Always false — explicit Complete/Abandon
EventDeserializer separate class	Inline switch + typed HandleAsync<T>
StopProcessingAsync in StopAsync	Use CloseAsync in StopAsync
await on StartProcessingAsync	Fire-and-forget: _processor.StartProcessingAsync(ct);
BackgroundService for DLQ	Paired ServiceBusProcessor in same listener
Missing LogContext enrichment	Push EventType, SessionId, MessageId via PropertyEnricher

Detect Existing Patterns

# Find ServiceBusSessionProcessor usage
grep -r "ServiceBusSessionProcessor\|CreateSessionProcessor" --include="*.cs" Infra/

# Find IHostedService listeners
grep -r "IHostedService" --include="*.cs" Infra/Services/ServiceBus/

# Find subject routing
grep -r "subject switch" --include="*.cs" Infra/Services/ServiceBus/

# Find dead letter processor
grep -r "SubQueue.DeadLetter" --include="*.cs" Infra/

Adding to Existing Project

1. Match the exact processor options shown above (PrefetchCount=1, etc.)
2. Add new event types to the HandleSubject switch expression
3. Add EventType constant for each new event in the constants class
4. DLQ processor reuses the same HandleSubject routing logic
5. Register via services.AddHostedService<YourListener>()
6. Serilog enrichment — push EventType, SessionId, MessageId for each message