Shuttle.Recall 22.0.0-beta.1

Shuttle.Recall

Shuttle.Recall is an event-sourcing mechanism for .NET that provides a flexible way to persist and retrieve event streams.

Installation

dotnet add package Shuttle.Recall

Registration

To register Shuttle.Recall, use the AddRecall extension method:

services.AddRecall(builder => 
{
    builder.AddProjection("ProjectionName", projection =>
    {
        projection.AddEventHandler<SomeEvent>((context, evt) => 
        {
            // handle event
        });
    });
});

The following types are registered:

• IEventStore (Scoped): Used to retrieve and save event streams.
• IEventProcessor (Singleton): Used to process projections.
• IEventMethodInvoker (Singleton): Invokes event handling methods on aggregate roots.
• ISerializer (Singleton): Serializes and deserializes events.
• IConcurrencyExceptionSpecification (Singleton): Detects concurrency exceptions.

Configuration Options

services.AddRecall(options =>
{
    options.EventProcessing.ProjectionThreadCount = 5;
    options.EventProcessing.IncludedProjections.Add("ProjectionName");
    options.EventProcessing.ExcludedProjections.Add("ExcludeMe");

    options.EventProcessing.ImmediateConsistency.Enabled = true;
    options.EventProcessing.ImmediateConsistency.IncludedProjections.Add("ProjectionName");

    options.EventStore.CompressionAlgorithm = "gzip";
    options.EventStore.EncryptionAlgorithm = "aes";
});

EventProcessing Options

ImmediateConsistency Options

By default, a projection only ever processes an event once the background IEventProcessor gets round to it. Immediate consistency lets specific projections handle an event synchronously, as part of the IEventStore.SaveAsync call that persisted it, so that a read model built from that projection is guaranteed to reflect the event by the time SaveAsync returns.

These IncludedProjections/ExcludedProjections only decide which projections are handled immediately — they are separate from, and have no effect on, the EventProcessing.IncludedProjections/ExcludedProjections above, which decide which projections the eventual IEventProcessor handles at all.

If a projection's handler throws while being invoked immediately, the event is not lost: EventProcessing.ImmediateConsistencyFailed is raised, and the eventual IEventProcessor will still pick up and retry the event on its next pass.

options.EventProcessing.ImmediateConsistencyFailed += (args, cancellationToken) =>
{
    _logger.LogWarning("Projection '{ProjectionName}' failed to handle event '{EventId}' immediately: {Exception}", args.ProjectionName, args.PrimitiveEvent.EventId, args.Exception);

    return Task.CompletedTask;
};

You can also request immediate consistency for a single SaveAsync call, regardless of whether Enabled is set — see Saving with Immediate Consistency.

EventStore Options

Usage

Saving an Event Stream

var eventStore = serviceProvider.GetRequiredService<IEventStore>();
var streamId = Guid.NewGuid();
var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream);

Saving with Headers

var stream = await eventStore.GetAsync(streamId, builder =>
{
    builder.AddHeader("key", "value");
});

stream.Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream);

Saving with Concurrency Check

var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

stream.ConcurrencyInvariant(5); // throws EventStreamConcurrencyException if version != 5

await eventStore.SaveAsync(stream);

Saving with Correlation ID

var stream = await eventStore.GetAsync(streamId);

stream
    .WithCorrelationId(correlationId)
    .Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream);

Saving with Immediate Consistency

var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream, builder =>
{
    builder.WithImmediateConsistency();
});

This requests immediate consistency for this save only, even when EventProcessing.ImmediateConsistency.Enabled is false. Which projections actually run immediately is still governed by EventProcessing.ImmediateConsistency.IncludedProjections/ExcludedProjections — see ImmediateConsistency Options.

Retrieving an Event Stream

var stream = await eventStore.GetAsync(streamId);

// Apply committed events to an aggregate root or state object
stream.Apply(someAggregateRoot);

Retrieving Events by Type

var stream = await eventStore.GetAsync(streamId);

// Get only committed events
var committedEvents = stream.GetEvents(EventStream.EventRegistrationType.Committed);

// Get only appended events
var appendedEvents = stream.GetEvents(EventStream.EventRegistrationType.Appended);

// Get all events
var allEvents = stream.GetEvents(EventStream.EventRegistrationType.All);

Committing Events Before Apply

var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

// Events are only applied after commit
stream.Commit();

// Now Apply() will include the committed events
stream.Apply(someAggregateRoot);

Removing an Event Stream

var stream = await eventStore.GetAsync(streamId);

stream.Remove();

await eventStore.RemoveAsync(streamId);

Projections

Handler Implementation

Implement the IEventHandler<T> interface to handle events:

public class OrderProjection : IEventHandler<OrderPlaced>
{
    public async Task HandleAsync(IEventHandlerContext<OrderPlaced> context, CancellationToken cancellationToken = default)
    {
        var evt = context.Event;
        var projection = context.Projection;
        var primitiveEvent = context.PrimitiveEvent;

        // Process the event
        await SaveToReadModelAsync(evt.OrderId, evt.Amount, cancellationToken);

        // Optionally defer for retry
        // context.Defer(TimeSpan.FromSeconds(5));
    }
}

Registering Projections

services.AddRecall(builder => 
{
    builder.AddProjection("OrderProjection", projection =>
    {
        projection.AddEventHandler<OrderProjection>();
    });
});

Inline Projection Handlers

services.AddRecall(builder => 
{
    builder.AddProjection("OrderProjection", projection =>
    {
        projection.AddEventHandler((IEventHandlerContext<OrderPlaced> context) =>
        {
            var evt = context.Event;
            // handle event inline
        });
    });
});

Delegate-based Handlers

builder.AddProjection("ProjectionName", (IEventHandlerContext<SomeEvent> context) =>
{
    // handle event
});

IEventProcessor Lifecycle

var processor = serviceProvider.GetRequiredService<IEventProcessor>();

await processor.StartAsync();

// ... application runs ...

await processor.StopAsync();

EventEnvelope Properties

The EventEnvelope class contains metadata about each event:

EventStream Properties

Exceptions

• EventStreamConcurrencyException: Thrown when concurrent modification is detected
• EventProcessingException: Thrown during projection event processing failures

Documentation

Please visit the Shuttle.Recall documentation for more information.