Generated Code¶

The Event Queue generator produces runtime types that handle channel management, event dispatch, and lifecycle — all backed by plain .NET APIs.

Runtime Types¶

ChannelWorker<T>¶

A BackgroundService base class that continuously reads from a Channel<T> and dispatches events to handlers.

EventQueueChannel<T>¶

Typed wrapper around Channel<T> providing three enqueue strategies:

EventAcknowledgement¶

A TaskCompletionSource wrapper for acknowledgement patterns. Use when a producer needs confirmation that an event was processed.

// Producer side
from ack in OrderEvents.EnqueueWithAck<AppRuntime>(new OrderPlaced("ord-1", "cust-1"))
from _ in liftEff(async () => await ack.WaitAsync())
select unit;

Deferred Transport¶

When a transport is set via SetTransport() before the event queue channel is initialized, the transport is stored as pending and applied automatically when Initialize() runs. This makes SetTransport call-order independent:

// Generated pattern (simplified)
private static IEventTransport<object>? _pendingTransport;

public static void SetTransport(IEventTransport<object> transport)
{
    if (_channel is not null)
        _channel.Transport = transport;      // channel exists — set directly
    else
        _pendingTransport = transport;       // channel not ready — defer
}

internal static void Initialize(EventQueueChannel<object> channel)
{
    _channel = channel;
    if (_pendingTransport is not null)
    {
        _channel.Transport = _pendingTransport;
        _pendingTransport = null;
    }
}

This is particularly important for subscriber event handlers ([EventQueueHandler]), where the handler's channel may not be initialized when SubscribeFromServiceBus<T>() calls SetTransport in the bootstrapper's configure callback.

Distributed Tracing with WithActivity¶

The generated ChannelWorker wraps each handler dispatch in a .WithActivity(...) call, producing System.Diagnostics.Activity spans for distributed tracing:

// Generated dispatch (simplified)
await OrderEventHandlers.Handle(e)
    .WithActivity("orders.Handle", ActivitySource, destination: "orders")
    .RunAsync(runtime);

Each queue class declares its own ActivitySource:

private static readonly ActivitySource ActivitySource =
    new("MyNamespace.OrderEvents", "1.0.0");

The activity name follows the pattern {queueName}.{methodName} and the destination tag is set to the queue name. These spans integrate with OpenTelemetry and any ActivityListener for end-to-end event flow visibility.

Enqueue Strategies¶

Fire-and-Forget¶

// Producer doesn't wait for processing
OrderEvents.Enqueue<AppRuntime>(new OrderPlaced("ord-1", "cust-1"));

Acknowledged¶

// Producer gets a handle to track completion
from ack in OrderEvents.EnqueueWithAck<AppRuntime>(new OrderPlaced("ord-1", "cust-1"))
select ack; // caller can await ack.WaitAsync() later

Synchronous Wait¶

// Producer blocks until processing completes
OrderEvents.EnqueueAndWait<AppRuntime>(new OrderPlaced("ord-1", "cust-1"));
// guaranteed processed when this completes

Concurrency Control¶

When MaxConcurrency > 1, the generated ChannelWorker<T> uses a SemaphoreSlim to limit the number of events being processed in parallel. The semaphore is acquired before dispatching to a handler and released when the handler completes (or faults).

Retry Policy¶

The generated worker implements a layered retry strategy controlled by the MaxRetries attribute property:

1. Immediate retries — failed dispatches are retried in-memory with exponential backoff (200ms, 400ms, 800ms...)
2. Delayed retries — after immediate retries are exhausted, the message is abandoned (re-enqueued) for transport-level retry with its own backoff
3. Dead-letter — after all retries are exhausted, the message is moved to the dead-letter queue

Attempt 1 (initial)    → fail → immediate retry 1 (200ms)
                                → fail → immediate retry 2 (400ms)
                                         → fail → immediate retry 3 (800ms)
                                                  → fail → abandon (transport backoff ~100ms)
Attempt 5 (redelivery) → fail → immediate retries 1-3...
                                → fail → abandon (transport backoff ~200ms)
Attempt 6 (redelivery) → fail → immediate retries 1-3...
                                → fail → dead-letter

The OnError(T, Exception) virtual method is called after immediate retries are exhausted. If overridden to return MessageAction.Retry or MessageAction.Skip, that action takes precedence over the default delayed retry / dead-letter path.

Timeout Enforcement¶

When TimeoutMilliseconds > 0 is set on the attribute, each dispatch attempt is wrapped with a linked CancellationTokenSource.CancelAfter(). Timeouts produce a TimeoutException that enters the retry pipeline. Shutdown cancellation is distinguished from timeout cancellation — only handler-initiated timeouts trigger retries.

Health Reporting¶

The generated worker exposes queue depth, processing rate, and error count. These can be wired into ASP.NET Core health checks or any monitoring system.

Graceful Shutdown¶

On cancellation, the worker stops reading new events and awaits all in-flight concurrent tasks via Task.WhenAll before stopping. In-flight tasks are tracked in a ConcurrentDictionary and removed on completion. The host's ShutdownTimeout acts as the overall deadline.

Testing¶

Event queues expose test helpers for verifying event flow in tests:

// Read a single event from the channel (non-blocking)
var evt = channel.TryRead();

// Drain all pending events
var allEvents = channel.DrainAll();

These helpers are available on the EventQueueChannel<T> instance, typically resolved from the test DI container.

Request-Response Transport¶

The Deepstaging.RequestResponse namespace provides a bidirectional transport abstraction for request-response patterns — in contrast to IEventTransport<TEvent> which is unidirectional fire-and-forget.

IRequestResponseTransport<TRequest, TResponse>¶

Correlation is handled internally by the transport — callers do not manage correlation IDs directly.

PendingRequest<TRequest, TResponse>¶

Represents a request awaiting a response. The receiver calls Reply(TResponse) to send the response or Fault(Exception) to signal failure.

ChannelRequestResponseTransport<TRequest, TResponse>¶

The default in-process implementation backed by System.Threading.Channels.Channel<T>. Supports both bounded and unbounded modes:

// Unbounded
var transport = new ChannelRequestResponseTransport<MyRequest, MyResponse>();

// Bounded with capacity
var transport = new ChannelRequestResponseTransport<MyRequest, MyResponse>(capacity: 100);

Both constructors accept optional singleReader (default true) and singleWriter (default false) parameters for channel optimization.

External transports (Azure Service Bus sessions, AMQP reply queues, gRPC, etc.) implement IRequestResponseTransport<TRequest, TResponse> to provide cross-process or cross-service request-response semantics.

Direct Usage¶

The generated types can be injected and used directly without the effects layer:

public class OrderService(EventQueueChannel<OrderEvent> channel)
{
    public async Task PlaceOrder(OrderPlaced evt) =>
        await channel.Enqueue(evt);
}