Observers

Observers are a lightweight alternative to middleware for cross-cutting concerns. They receive lifecycle notifications at fixed points (before, after, on error) without needing to call next. Use them when you want to observe the pipeline, not intercept it.

Exceptions thrown by observers are caught and logged individually. A failing observer never affects other observers or the message pipeline.

IConsumeObserver

Receives notifications for every inbound message across all transport consumers:

public class AuditConsumeObserver(IAuditLog audit) : IConsumeObserver
{
    public Task OnConsumingAsync<T>(ConsumeContext<T> context)
    {
        audit.Record($"Consuming {typeof(T).Name} at {context.Timestamp}");
        return Task.CompletedTask;
    }

    public Task OnConsumedAsync<T>(ConsumeContext<T> context)
    {
        audit.Record($"Consumed {typeof(T).Name}: success");
        return Task.CompletedTask;
    }

    public Task OnConsumeErrorAsync<T>(ConsumeContext<T> context, Exception exception)
    {
        audit.Record($"Consume error: {exception.Message}");
        return Task.CompletedTask;
    }
}

All three methods have default implementations that return Task.CompletedTask, so you only need to implement what you care about.

Register at the global level:

emit.AddConsumeObserver<AuditConsumeObserver>();

IConsumeObserver fires for Kafka consumers. It does not fire for mediator requests; use IMediatorObserver for those.

IProduceObserver

Fires for every produce call through the outbound pipeline.

Method	When it fires
OnProducingAsync<T>(SendContext<T>)	Before the outbound pipeline executes
OnProducedAsync<T>(SendContext<T>)	Produce succeeded
OnProduceErrorAsync<T>(SendContext<T>, Exception)	Pipeline threw before or during the outbox write

Register globally:

emit.AddProduceObserver<MyProduceObserver>();

IOutboxObserver

Fires at outbox entry lifecycle points: enqueue and the worker’s processing stages.

Method	When it fires
OnEnqueuedAsync	Entry was written to the outbox (fires from the outbound pipeline)
OnProcessingAsync	Worker is about to attempt delivery
OnProcessedAsync	Delivery succeeded; entry will be deleted
OnProcessErrorAsync	Delivery failed; entry stays for the next poll cycle

Register globally:

emit.AddOutboxObserver<MyOutboxObserver>();

OnEnqueuedAsync fires in your request scope. The remaining three fire from the background worker.

IMediatorObserver

Fires for mediator request/response operations.

Method	When it fires
OnHandlingAsync<T>(MediatorContext<T>)	Before the handler is invoked
OnHandledAsync<T>(MediatorContext<T>)	After the handler completes successfully
OnHandleErrorAsync<T>(MediatorContext<T>, Exception)	If the handler throws

IMediatorObserver is registered on the mediator builder, not on EmitBuilder:

emit.AddMediator(mediator =>
{
    mediator.AddObserver<MyMediatorObserver>();
});

IDaemonObserver

Fires as daemons are assigned, started, stopped and moved.

Method	When it fires
OnDaemonAssignedAsync	The leader assigned a daemon to a node (fires on the leader)
OnDaemonStartedAsync	The daemon has started running on this node
OnDaemonStoppedAsync	The daemon stopped on this node, including when it failed or the node shuts down
OnDaemonRevokedAsync	The leader revoked a daemon from a node (fires on the leader)

All four methods receive (string daemonId, Guid nodeId, CancellationToken).

Register globally:

emit.AddDaemonObserver<MyDaemonObserver>();
emit.AddLeaderElectionObserver<MyLeaderElectionObserver>();

Daemon and leader election observers are notified in order, away from the heartbeat, so a slow observer delays later notifications but never coordination.

Use OnDaemonRevokedAsync to see when a daemon moves to another node.

Observer vs. middleware

Observers are the right tool when you need simple before/after/error hooks and do not need to:

• Intercept or modify the message
• Short-circuit the pipeline
• Execute async code at a specific position relative to other middleware

Use middleware when you need any of the above, or when execution order matters relative to other pipeline stages.