fullstackhero

Reference

Eventing building block

Event bus implementation (InMemory or RabbitMQ), outbox/inbox stores for durable delivery, and the open-generic handler registration that lets modules subscribe with one method.

views 0 Last updated

The Eventing block is the runtime side of the kit’s integration-event story. It ships two IEventBus implementations - InMemoryEventBus (synchronous, in-process) and RabbitMqEventBus (durable, cross-service) - an EF Core-backed outbox store for transactional delivery, an inbox store for receiver-side idempotency, a JsonEventSerializer, and the auto-discovery hook that registers every IIntegrationEventHandler<TEvent> in your module assemblies.

What it ships

Extensions

  • AddEventingCore(services, configuration) - called once by the host, not per module. Registers JsonEventSerializer, a no-op IEventTenantScope default (the Multitenancy module swaps in a Finbuckle-backed one), the EventingDbContext that owns the outbox/inbox tables, exactly one IOutboxStore / IInboxStore / IOutboxWriter, the OutboxDispatcher, an IDbInitializer for the framework schema, and the single-database drain defaults. Picks the IEventBus from EventingOptions:Provider (InMemory default, RabbitMQ) and adds the dispatcher hosted service when UseHostedServiceDispatcher is true.
  • AddIntegrationEventHandlers(services, assemblies[]) - scans the supplied assemblies for IIntegrationEventHandler<TEvent> implementations and registers them scoped in DI.

Event bus implementations

  • InMemoryEventBus - in-process. For each event it sets the tenant context first (IEventTenantScope.Begin(event.TenantId)), creates a fresh DI scope, resolves the matching handlers, checks the inbox (skip if already processed by that handler), awaits each handler in order, then marks the inbox row. Great for dev/test and single-process production hosts.
  • RabbitMqEventBus - durable, cross-service; uses RabbitMQ.Client and publishes to a durable topic exchange (EventingOptions:RabbitMQ).

Outbox / inbox

  • EventingDbContext - the single, framework-owned context that maps OutboxMessages and InboxMessages into schema framework. It derives from BaseDbContext, so it follows the tenant connection: a tenant with a dedicated database gets its outbox rows in that database, beside the business data they accompany.
  • IOutboxWriter (AddAsync) - the publish-side contract, in Eventing.Abstractions. This is what a module injects; it never touches the eventing runtime.
  • IOutboxStore (IOutboxWriter + ClaimBatchAsync, MarkAsProcessedAsync, MarkAsFailedAsync, GetDeadLetteredAsync, RedriveDeadLettersAsync) + EfCoreOutboxStore - the dispatcher-side surface. AddAsync joins the caller’s transaction when one is open, so the event commits or rolls back with the business write; with no transaction it commits on its own.
  • IInboxStore + EfCoreInboxStore - dedupe table keyed by (event id, handler name), so each handler processes an event at most once.
  • OutboxDispatcher - scoped service; claims a batch (OutboxBatchSize, default 100) with a lease, deserializes, publishes via IEventBus, marks processed. A failing row increments RetryCount and backs off exponentially (NextRetryAt); after OutboxMaxRetries (default 5) it’s flagged IsDead.
  • OutboxDispatcherHostedService - background loop calling the dispatcher every OutboxDispatchIntervalSeconds (default 10), once per drain target (see below).
  • OutboxMessage - Id, CreatedOnUtc, Type, Payload, TenantId, CorrelationId, ProcessedOnUtc, RetryCount, LastError, IsDead, NextRetryAt, ClaimedUntilUtc, ClaimedBy. Implements IGlobalEntity (background processors must scan across tenants).
  • InboxMessage - Id + HandlerName (composite key), EventType, ProcessedOnUtc, TenantId. Also IGlobalEntity.

Serializer

  • JsonEventSerializer - System.Text.Json; the outbox stores the event’s type name alongside the payload so the dispatcher can rehydrate the concrete event for publishing.

Drain targets and multi-instance safety

Because outbox rows follow the tenant connection, the dispatcher can’t poll a single database. Each cycle it asks IEventingDrainTargetProvider which databases to drain and runs one pass per target inside IEventingDrainScope, which installs the tenant context before the scope builds its EventingDbContext (that context captures TenantInfo, and with it the connection string, at construction). BuildingBlocks defaults to a single target; the Multitenancy module replaces it with the default database plus one target per distinct active per-tenant connection string - tenants sharing a database collapse to one pass. An unreachable tenant database is logged and skipped, never fatal to the cycle.

ClaimBatchAsync leases rows with Postgres FOR UPDATE SKIP LOCKED in one UPDATE … RETURNING, so several API instances partition a batch instead of all publishing the same message. The lease is an expiry (OutboxClaimLeaseSeconds, default 300), so a dispatcher that dies mid-batch has its rows recovered rather than stranded. Providers without SKIP LOCKED fall back to an unclaimed read and log a warning that only one instance is safe.

How modules consume Eventing

Publishing needs no module registration at all - the host calls AddEventingCore once. A module that handles events registers its handlers:

public void ConfigureServices(IHostApplicationBuilder builder)
{
    builder.Services.AddHeroDbContext<TicketsDbContext>();
    builder.Services.AddIntegrationEventHandlers(typeof(TicketsModule).Assembly);
}

Publish through the outbox. Inject IOutboxWriter and add the event next to your business change; the dispatcher publishes it on the next cycle:

public async ValueTask<Unit> Handle(ResolveTicketCommand cmd, CancellationToken ct)
{
    var ticket = await _db.Tickets.FindAsync([cmd.TicketId], ct).ConfigureAwait(false);
    ticket!.Resolve(cmd.ResolutionNote);
    await _db.SaveChangesAsync(ct).ConfigureAwait(false);


    await _outbox.AddAsync(new TicketResolvedIntegrationEvent(/* … */), ct).ConfigureAwait(false);
    return Unit.Value;
}

Delivery is asynchronous. The consumer runs in the dispatcher’s scope on the next cycle, not inside your request - so its failures never reach your caller, and no caller (or test) should assume the side effect has already happened.

Calling IEventBus.PublishAsync directly runs handlers immediately, in-process, but the event is neither durable nor transactional. The kit does this in exactly one place: Chat mention notifications, where the handler pushes over SignalR and a delayed badge reads as broken. Prefer the outbox everywhere else.

The OutboxDispatcher picks up pending rows on the next interval (or you can call DispatchAsync from a Hangfire job if you don’t want a hosted-service loop). The receiving side declares a handler:

public sealed class TicketResolvedNotifyHandler(/* … */)
    : IIntegrationEventHandler<TicketResolvedIntegrationEvent>
{
    public async Task HandleAsync(TicketResolvedIntegrationEvent evt, CancellationToken ct = default)
    {
        // write a notification, send an email, etc.
    }
}

Host registers handlers in bulk via AddIntegrationEventHandlers against all module marker assemblies:

builder.Services.AddIntegrationEventHandlers(moduleAssemblies);

Configuration

{
  "EventingOptions": {
    "Provider": "InMemory",                       // or "RabbitMQ"
    "OutboxBatchSize": 100,
    "OutboxMaxRetries": 5,
    "OutboxRetryBaseDelaySeconds": 30,            // exponential backoff after a failure
    "OutboxRetryMaxDelaySeconds": 3600,
    "OutboxClaimLeaseSeconds": 300,               // must exceed worst-case time to publish a batch
    "EnableInbox": true,
    "OutboxDispatchIntervalSeconds": 10,
    "UseHostedServiceDispatcher": true,
    "RabbitMQ": {
      "Host": "rabbitmq",
      "Port": 5672,
      "UserName": "guest",
      "Password": "guest",
      "VirtualHost": "/",
      "ExchangeName": "fsh.events",
      "QueuePrefix": "fsh",
      "UseSsl": false,
      "PublishRetryCount": 3,
      "PublishRetryDelayMs": 1000
    }
  }
}

Set UseHostedServiceDispatcher = false when you’d rather drive the dispatcher from Hangfire on a fixed schedule (more deterministic for some ops setups).

How to extend

Add another transport

Implement IEventBus; register your implementation in place of InMemoryEventBus / RabbitMqEventBus. Bus consumers don’t care which one is wired.

Add a side-channel like Outbox-to-Kafka

OutboxDispatcher is small and replaceable. Subclass or wrap it to publish to Kafka in addition to RabbitMQ, or to write to multiple destinations.

Skip the outbox for cheap fire-and-forget

IEventBus.PublishAsync still goes straight to the bus. If an event genuinely needs latency over durability (a cache-invalidation hint, a metrics ping, a realtime badge), publish it directly - but leave a comment saying why, because the kit’s default is the outbox and a reviewer will ask.

Drain a different set of databases

Replace IEventingDrainTargetProvider (and IEventingDrainScope if the target needs more than a tenant context installed) to teach the dispatcher about databases the tenant store doesn’t know - a read replica, an archive shard, a per-region cluster.

Gotchas

  • Domain events vs integration events are different things in the kit. Domain events fire inside the SaveChanges interceptor (synchronous, in-module, via Mediator). Integration events go through the outbox and IEventBus (asynchronous, cross-module / cross-service). Use the right one - domain events for invariants and bookkeeping inside the module, integration events for everything else.
  • Outbox dispatcher is scoped per cycle. Each poll gets a fresh DbContext scope; a failing publish increments the row’s RetryCount (with LastError recorded) and backs off before the next attempt. At OutboxMaxRetries the row is flagged IsDead - monitor it (outbox.deadlettered) and replay with RedriveDeadLettersAsync; dead rows don’t retry themselves.
  • One store, owned by the framework. OutboxMessages/InboxMessages belong to EventingDbContext in schema framework, never to a module’s context. Registering a second IOutboxStore makes .NET DI resolve whichever one registered last for the whole application - the defect behind #1349, where a second module publishing broke every module’s outbox. EventingRegistrationTests guards the count.
  • OutboxClaimLeaseSeconds must exceed the worst-case time to publish a batch. If a batch can outlive its lease, another instance re-claims rows still in flight and publishes them twice.
  • A background publisher must set the tenant context before AddAsync, not just before the publish - with per-tenant databases the row otherwise lands in the wrong database and is drained under the wrong target.
  • Inbox dedupes per handler, by (EventId, HandlerName). A redelivered event is skipped only for handlers that already completed it; if you mint two events with the same Id, the second is silently dropped for every handler. Always generate a fresh Guid per event.
  • Background publishers need the tenant context set. InMemoryEventBus does this for you via IEventTenantScope (reading event.TenantId) before resolving handlers - a MultiTenantDbContext captures its tenant at construction, so setting it later is too late. If you build your own bus or dispatch path, preserve this ordering or tenant-filtered handlers NRE.
  • RabbitMQ publishing retries in-process (PublishRetryCount / PublishRetryDelayMs), but consumers own their durability via inbox + retry semantics.

Critical files

  • src/BuildingBlocks/Eventing/ServiceCollectionExtensions.cs
  • src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs
  • src/BuildingBlocks/Eventing/RabbitMq/RabbitMqEventBus.cs
  • src/BuildingBlocks/Eventing/Outbox/OutboxDispatcher.cs
  • src/BuildingBlocks/Eventing/Inbox/EfCoreInboxStore.cs