Hosted Service — ServiceBusSessionProcessor Lifecycle

Core Principles

• Each listener implements IHostedService directly (not BackgroundService)
• Constructor creates BOTH a ServiceBusSessionProcessor AND a paired ServiceBusProcessor for the dead letter sub-queue
• AutoCompleteMessages = false -- always complete/abandon explicitly
• Each ServiceBus client is a typed singleton wrapper around ServiceBusClient
• StartAsync starts both processors; StopAsync closes both
• Error handling via ProcessErrorAsync callback with LogCritical

Key Patterns

IHostedService Listener with Paired DLQ Processor

namespace {Company}.{Domain}.Processor.Infra.ServiceBus.Listeners;

public class OrderListener : IHostedService
{
    private readonly ILogger<OrderListener> _logger;
    private readonly IServiceProvider _provider;
    private readonly ServiceBusSessionProcessor _processor;
    private readonly ServiceBusProcessor _deadLetterProcessor;

    public OrderListener(
        IServiceProvider serviceProvider,
        ILogger<OrderListener> logger,
        IOptions<ServiceBusOptions> options,
        OrderServiceBus orderServiceBus)
    {
        var serviceBusOptions = options.Value;

        _logger = logger;
        _provider = serviceProvider;

        // Main session processor for ordered event processing
        _processor = orderServiceBus.Client.CreateSessionProcessor(
            serviceBusOptions.OrderCommandTopic,
            serviceBusOptions.Subscription,
            new ServiceBusSessionProcessorOptions()
            {
                PrefetchCount = 1,
                MaxConcurrentCallsPerSession = 1,
                MaxConcurrentSessions = 1000,
                AutoCompleteMessages = false,
                SessionIdleTimeout = TimeSpan.FromMinutes(1),
                MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5),
            });

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

        // Paired DLQ processor for reprocessing dead-lettered messages
        _deadLetterProcessor = orderServiceBus.Client.CreateProcessor(
            serviceBusOptions.OrderCommandTopic,
            serviceBusOptions.Subscription,
            new ServiceBusProcessorOptions()
            {
                PrefetchCount = 1,
                AutoCompleteMessages = false,
                MaxConcurrentCalls = 1,
                SubQueue = SubQueue.DeadLetter,
            });

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

    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,
            "OrderListener => Processor_ProcessErrorAsync " +
            "Message handler encountered an exception," +
            " Error Source:{ErrorSource}," +
            " Entity Path:{EntityPath}",
            arg.ErrorSource.ToString(),
            arg.EntityPath);

        return Task.CompletedTask;
    }

    private async Task DeadLetterProcessor_ProcessMessageAsync(ProcessMessageEventArgs arg)
    {
        // DLQ handler uses the SAME routing logic as the main processor
        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 routing and HandleAsync defined in event-routing skill
    private Task<bool> HandleSubject(string subject, ServiceBusReceivedMessage message)
    {
        // See event-routing SKILL for full pattern
        return subject switch
        {
            EventType.OrderCreated => HandleAsync<OrderCreatedData>(message),
            EventType.OrderUpdated => HandleAsync<OrderUpdatedData>(message),
            _ => Task.FromResult(true) // Complete unknown events
        };
    }

    private async Task<bool> HandleAsync<T>(ServiceBusReceivedMessage message) { /* see event-routing */ return true; }

    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;
    }
}

Typed ServiceBusClient Wrapper

namespace {Company}.{Domain}.Processor.Infra.ServiceBus.Clients;

// Each external Service Bus namespace gets a typed wrapper
public class OrderServiceBus(string connectionString)
{
    public ServiceBusClient Client { get; } = new ServiceBusClient(connectionString);
}

ServiceBusOptions

namespace {Company}.{Domain}.Processor.Infra.ServiceBus;

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

    [Required]
    public required string OrderCommandTopic { get; init; }
    [Required]
    public required string Subscription { get; init; }
    public int BatchSize { get; init; } = 50;
    public int MaxConcurrentSessions { get; init; } = 100;
}

Registration

// In ServiceBusRegistrationExtensions.cs
services.AddSingleton(provider =>
{
    var options = provider.GetRequiredService<IOptions<ConnectionStringsOption>>();
    return new OrderServiceBus(options.Value.OrderServiceBus);
});

services.AddHostedService<OrderListener>();
services.AddHostedService<ProductListener>();

Exact Session Processor Configuration

Option	Value	Purpose
PrefetchCount	1	Conservative prefetch for reliability
MaxConcurrentCallsPerSession	1	Strict ordering within each session
MaxConcurrentSessions	1000	High parallelism across different aggregates
AutoCompleteMessages	false	Explicit complete/abandon for error control
SessionIdleTimeout	1 min	Release session lock after idle period
MaxAutoLockRenewalDuration	5 min	Renew lock for long-running handlers

Exact DLQ Processor Configuration

Option	Value	Purpose
PrefetchCount	1	Process one DLQ message at a time
AutoCompleteMessages	false	Explicit complete/abandon
MaxConcurrentCalls	1	Serial DLQ processing
SubQueue	SubQueue.DeadLetter	Target the dead letter sub-queue

Anti-Patterns

Anti-Pattern	Correct Approach
Creating processor in StartAsync	Create in constructor, start in StartAsync
Missing DLQ processor	Always pair main processor with DLQ processor
AutoCompleteMessages = true	Set false for explicit control
Using StopProcessingAsync	Use CloseAsync in StopAsync
Missing error handler	Always wire ProcessErrorAsync with LogCritical

Detect Existing Patterns

# Find IHostedService implementations
grep -r ": IHostedService" --include="*.cs" src/

# Find ServiceBusSessionProcessor usage
grep -r "ServiceBusSessionProcessor" --include="*.cs" src/

# Find SubQueue.DeadLetter usage
grep -r "SubQueue.DeadLetter" --include="*.cs" src/

# Find AddHostedService registrations
grep -r "AddHostedService" --include="*.cs" src/

Adding to Existing Project

1. Create typed ServiceBus client wrapper in Infra/ServiceBus/Clients/
2. Add connection string to ConnectionStringsOption
3. Add topic/subscription names to ServiceBusOptions
4. Create listener class implementing IHostedService with paired DLQ processor
5. Register singleton client and AddHostedService<T>() in DI extension
6. Match exact configuration values: PrefetchCount=1, MaxConcurrentSessions=1000, etc.