faysilalshareef/event-routing

Event Routing — Subject-Based Inline Dispatch

Core Principles

• Routing is INLINE in each listener class (no separate IEventRouter or EventDeserializer)
• HandleSubject() method uses a switch expression on message.Subject
• Each case calls a typed HandleAsync<TEventData>(message) method
• HandleAsync<T> deserializes Event<T> from JSON body, creates a DI scope, and calls IMediator.Send()
• Returns bool: true = complete message, false = abandon message
• Unknown subjects return Task.FromResult(true) to complete (not retry)
• LogContext.Push with PropertyEnricher for EventType, SessionId, MessageId

Key Patterns

Subject-Based Switch in Listener

// Inside the listener class (e.g., OrderListener)
private Task<bool> HandleSubject(string subject, ServiceBusReceivedMessage message)
{
    return subject switch
    {
        EventType.OrderCreated => HandleAsync<OrderCreatedData>(message),
        EventType.OrderUpdated => HandleAsync<OrderUpdatedData>(message),
        _ => Task.FromResult(true) // Complete unknown events
    };
}

Event Type Constants

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

public class EventType
{
    public const string OrderCreated = "OrderCreated";
    public const string OrderUpdated = "OrderUpdated";
    public const string OrderCompleted = "OrderCompleted";
}

// Separate constants class per source domain
public static class AccountEventType
{
    public const string AccountConfirmed = "AccountConfirmed";
    public const string AccountUpdated = "AccountUpdated";
    public const string SubscriptionCreated = "SubscriptionCreated";
    public const string SubscriptionRemoved = "SubscriptionRemoved";
}

HandleAsync with Serilog Enrichment

// Inside the listener class
private async Task<bool> HandleAsync<T>(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))
    {
        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 = _provider.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;
}

Event<T> Envelope

namespace {Company}.{Domain}.Processor.Events;

public class Event<T> : IRequest<bool>
{
    public required Guid AggregateId { get; init; }
    public required Guid? UserId { get; init; }
    public required int Sequence { get; init; }
    public required string Type { get; init; }
    public required T Data { get; init; }
    public required DateTime DateTime { get; set; }
    public required int Version { get; init; }

    public string GetDataAsJson() =>
        JsonConvert.SerializeObject(Data, new JsonSerializerSettings()
        {
            Formatting = Formatting.Indented,
            NullValueHandling = NullValueHandling.Ignore,
            ContractResolver = new CamelCasePropertyNamesContractResolver()
        });
}

ProcessMessageAsync Calls HandleSubject

// Main processor handler (ProcessSessionMessageEventArgs)
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);
}

// DLQ processor handler (ProcessMessageEventArgs -- different type, no session)
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);
}

Multi-Source Listener Example

// A listener consuming events from a different domain
private Task<bool> HandleSubject(string subject, ServiceBusReceivedMessage message)
{
    return subject switch
    {
        AccountEventType.AccountConfirmed => HandleAsync<AccountConfirmedData>(message),
        AccountEventType.AccountUpdated => HandleAsync<AccountUpdatedData>(message),
        AccountEventType.SubscriptionCreated => HandleAsync<SubscriptionCreatedData>(message),
        AccountEventType.SubscriptionRemoved => HandleAsync<SubscriptionRemovedData>(message),
        _ => Task.FromResult(true) // Ignore unknown events
    };
}

Complete Flow

ServiceBusSessionProcessor
  -> ProcessMessageAsync(ProcessSessionMessageEventArgs arg)
    -> HandleSubject(arg.Message.Subject, arg.Message)
      -> switch on Subject string
        -> HandleAsync<TEventData>(message)
          -> PropertyEnricher (EventType, SessionId, MessageId)
          -> LogContext.Push(...)
          -> Encoding.UTF8.GetString(message.Body)
          -> JsonConvert.DeserializeObject<Event<T>>(json)
          -> _provider.CreateScope()
          -> scope.GetRequiredService<IMediator>()
          -> mediator.Send(@event)
    -> if true: CompleteMessageAsync
    -> if false: AbandonMessageAsync

Anti-Patterns

| Anti-Pattern | Correct Approach | |---|---| | Separate EventRouter/EventDeserializer class | Inline HandleSubject + HandleAsync in each listener | | Using LogContext.PushProperty | Use LogContext.Push with PropertyEnricher instances | | Retrying unknown event types | Return Task.FromResult(true) for unknown subjects | | Getting body via message.Body.ToString() | Use Encoding.UTF8.GetString(message.Body) | | Using System.Text.Json | Use Newtonsoft.Json (JsonConvert.DeserializeObject) | | Creating mediator outside scope | Create DI scope, then resolve IMediator from scope |

Detect Existing Patterns

# Find subject-based routing
grep -r "Subject\|HandleSubject\|HandelSubject" --include="*.cs" src/

# Find HandleAsync methods
grep -r "HandleAsync<" --include="*.cs" src/

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

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

Adding to Existing Project

1. Add new event type constant to the appropriate constants class
2. Create event data class in Events/Incoming/{SourceDomain}/
3. Add case to HandleSubject switch in the relevant listener
4. Create MediatR handler implementing IRequestHandler<Event<TData>, bool>
5. Follow existing LogContext enrichment -- always push EventType, SessionId, MessageId