Documentation Index
Fetch the complete documentation index at: https://mintlify.com/Eventuous/eventuous/llms.txt
Use this file to discover all available pages before exploring further.
Subscriptions are the backbone of event-driven read-side processing in Eventuous. A subscription is a hosted service that continuously reads events from a stream or the global event log and dispatches each event through a configurable pipeline of filters before delivering it to one or more event handlers. Because they implement IHostedService, subscriptions start and stop automatically with your application and reconnect transparently after transient failures.
The IMessageSubscription interface
Every subscription type implements IMessageSubscription, which defines the three core members:
public interface IMessageSubscription {
string SubscriptionId { get; }
ValueTask Subscribe(
OnSubscribed onSubscribed,
OnDropped onDropped,
CancellationToken cancellationToken);
ValueTask Unsubscribe(
OnUnsubscribed onUnsubscribed,
CancellationToken cancellationToken);
}
SubscriptionId — unique identifier for this subscription instance; used for diagnostics, checkpointing, and DI keying.Subscribe — begins consuming events; fires the OnSubscribed delegate when the connection is established and OnDropped if it is lost.Unsubscribe — gracefully stops the subscription and fires OnUnsubscribed.
EventSubscription<T> base class
EventSubscription<T> is the abstract base class for all Eventuous subscriptions. The generic parameter T is constrained to SubscriptionOptions.
public abstract class EventSubscription<T> : IMessageSubscription, IAsyncDisposable
where T : SubscriptionOptions
| Member | Description |
|---|
SubscriptionId | Delegates to Options.SubscriptionId |
IsRunning | true while the subscription is active |
IsDropped | true after a connection drop, before reconnection |
Options | Typed options instance for this subscription |
Pipe | The ConsumePipe that events flow through |
IMessageConsumeContext, and calls Pipe.Send(context). The pipe processes the context through each registered filter in order, eventually reaching the consumer that invokes your handlers.
SubscriptionOptions
SubscriptionOptions is an abstract record from which all concrete option types inherit.
public abstract record SubscriptionOptions {
// Required: uniquely identifies this subscription
public string SubscriptionId { get; set; } = null!;
// When true, the subscription stops on any unhandled error
public bool ThrowOnError { get; set; }
}
SubscriptionWithCheckpointOptions, which adds checkpoint-tuning knobs:
public abstract record SubscriptionWithCheckpointOptions : SubscriptionOptions {
// Commit checkpoint after this many events (default: 100)
public int CheckpointCommitBatchSize { get; set; } = 100;
// Commit checkpoint after this delay in ms (default: 5 000)
public int CheckpointCommitDelayMs { get; set; } = 5000;
// Where to start if no checkpoint exists (default: Earliest)
public InitialPosition StartFrom { get; set; } = InitialPosition.Earliest;
}
EventSubscriptionWithCheckpoint<T>
For subscriptions that need durable progress tracking, Eventuous provides EventSubscriptionWithCheckpoint<T>. It extends EventSubscription<T> with:
CheckpointStore — an ICheckpointStore used to load and persist the last processed position.CheckpointCommitHandler — batches CommitPosition values and writes the checkpoint after either the configured batch size or delay is reached, whichever comes first.- Resubscribe safety — on reconnect,
LastProcessed is reset so the subscription resumes from the last committed checkpoint rather than from where it dropped.
The consume pipe
Events flow through a ConsumePipe — a linked list of IConsumeFilter stages — before reaching your handlers. Filters can transform, partition, or drop contexts. The built-in filters are:
| Filter | Purpose |
|---|
AsyncHandlingFilter | Offloads handler invocations to a bounded channel with configurable concurrency |
PartitioningFilter | Routes events to one of N per-partition AsyncHandlingFilter workers |
TracingFilter | Wraps each message in an OpenTelemetry activity |
ConsumerFilter | Delivers the context to the registered IMessageConsumer (final stage) |
SubscriptionBuilder.AddConsumeFilterFirst or AddConsumeFilterLast.
Registering a subscription
Use the AddSubscription<T, TOptions> extension on IServiceCollection to wire everything up with the DI container. The lambda receives a SubscriptionBuilder that provides a fluent API for configuring the checkpoint store, handlers, and pipe filters.
services.AddSubscription<AllStreamSubscription, AllStreamSubscriptionOptions>(
"BookingsProjections",
builder => builder
.UseCheckpointStore<MongoCheckpointStore>()
.AddEventHandler<BookingStateProjection>()
.AddEventHandler<MyBookingsProjection>()
.WithPartitioningByStream(2)
);
SubscriptionBuilder API
| Method | Description |
|---|
.Configure(Action<TOptions>) | Configure subscription-specific options |
.UseCheckpointStore<T>() | Override the checkpoint store for this subscription |
.AddEventHandler<T>() | Register an event handler (keyed singleton) |
.AddEventHandler<T>(Func<IServiceProvider, T>) | Register a handler with a custom factory |
.AddEventHandler<T>(T instance) | Register a pre-built handler instance |
.WithPartitioningByStream(int) | Enable per-stream partitioned concurrency |
.WithPartitioning(int, GetPartitionKey) | Enable partitioning with a custom key function |
.AddConsumeFilterFirst/Last(filter) | Insert a custom filter at the start or end of the pipe |
.UseConsumer(factory) | Replace the default DefaultConsumer |
.AddEventHandler<T>() is registered as a keyed singleton scoped to SubscriptionId, so multiple subscriptions can share the same handler type without conflicts.
Lifecycle as IHostedService
AddSubscription automatically registers a SubscriptionHostedService that adapts the subscription to .NET’s IHostedService contract. On StartAsync it calls Subscribe; on StopAsync it calls Unsubscribe. Health checks are registered automatically and can be wired into ASP.NET Core’s health-check infrastructure with AddSubscriptionsHealthCheck.