Interface IEventSource

Namespace

Stratara.Abstractions.EventSourcing

Assembly

Stratara.Abstractions.dll

Write-side façade for the event store. Command handlers append events through this service; the implementation tracks pending writes and flushes them on SaveChangesAsync(CancellationToken).

public interface IEventSource

Extension Methods

FieldChangedEventExtensions.ApplyPropertyChanged<TAggregate>(TAggregate, FieldChangedEvent<TAggregate>)

EventStream.ApplyEvents(object, IReadOnlyList<IEvent>)

Examples

Append events from a command handler and commit the unit-of-work:

public sealed class CreateOrderHandler(IEventSource events) : ICommandHandler<CreateOrder>
{
    public async Task HandleAsync(CreateOrder command, CancellationToken cancellationToken)
    {
        await events.CreateAsync<Order>(command.OrderId,
            new OrderCreated(command.OrderId, command.CustomerId, command.Amount),
            cancellationToken);
        await events.SaveChangesAsync(cancellationToken);
    }
}

Remarks

Every event is recorded under the tenant that owns it — its Subject — and that tenant is not simply the one in the caller's session. For each event the store takes the first of these that names a tenant: the Subject stated for that event with AppendOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken); the owner already resolved for the same stream earlier in the batch; the owner recorded on the stream's first event — its tenant, and its user where one was recorded; the TenantId of a creation event; and only then the tenant in the session. If none of them names a tenant, the append fails.

A stream therefore keeps the owner it was created with, user included, whatever session appends to it later; a stream whose first event names no user is given none. A first event that states no owner takes the tenant and the data-owner user in the session, so an aggregate created for another tenant states that tenant on its first event by implementing IAggregateCreationEvent. A creation event names no user, so a stream it creates has none.

Methods

AppendAsync<TAggregate>(Guid, object, CancellationToken)

Append an event to an existing stream.

Task AppendAsync<TAggregate>(Guid streamId, object @event, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

event object

The event payload.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type.

Remarks

The event takes the owner recorded on the stream, not the tenant in the session. On a stream that does not exist yet, the owner is resolved as for the first event of CreateAsync<TAggregate>(Guid, object, CancellationToken). Use AppendOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken) for an event whose owner differs from the stream's.

Exceptions

SessionRequiredException

No session context is set on the current scope.

AppendOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken)

Append an event with an explicit Subject (data owner), overriding every other source — the stream's recorded owner, a creation event's tenant and the session — for this one event.

Task AppendOnBehalfOfAsync<TAggregate>(Guid streamId, object @event, EventSubject subject, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

event object

The event payload.

subject EventSubject

The data owner to record. Its tenant id must be non-empty: stating the Subject also states that the stream, the event and the session are not to be consulted, so an empty one fails the append rather than falling back to them.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type.

Remarks

Use this method for an event whose owner differs from the stream's. The actor recorded with the event stays the caller's session; only the owner changes, and only for this event — the next append to the stream takes the stream's owner again, in the same batch as in a later one. On a stream's first event the stated owner is the one the stream records.

The event's protected fields are encrypted under the stated owner's keys, so an erasure of the stream's owner does not reach them, and an erasure of the stated owner can leave the stream unable to rehydrate.

Exceptions

ArgumentException

subject names no tenant.

SessionRequiredException

No session context is set on the current scope.

AppendRangeAsync<TAggregate>(Guid, IEnumerable<object>, CancellationToken)

Append multiple events to an existing stream in order.

Task AppendRangeAsync<TAggregate>(Guid streamId, IEnumerable<object> events, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

events IEnumerable<object>

The events to append, in order.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type.

Remarks

The events take the owner recorded on the stream, as in AppendAsync<TAggregate>(Guid, object, CancellationToken).

Exceptions

SessionRequiredException

No session context is set on the current scope.

CreateAsync<TAggregate>(Guid, object, CancellationToken)

Create a new stream with the first event. Fails if the stream already exists.

Task CreateAsync<TAggregate>(Guid streamId, object @event, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

event object

The creation event. Implement IAggregateCreationEvent on it to state the new stream's owner instead of taking the tenant in the session.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type the stream represents.

Remarks

The new stream's owner is the tenant the event carries when it is an IAggregateCreationEvent with a non-empty tenant, otherwise the tenant in the session. Every later event on the stream keeps that owner.

Exceptions

SessionRequiredException

No session context is set on the current scope.

CreateRangeAsync<TAggregate>(Guid, IEnumerable<object>, CancellationToken)

Create a new stream with multiple events in order.

Task CreateRangeAsync<TAggregate>(Guid streamId, IEnumerable<object> events, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

events IEnumerable<object>

The events to append, in order.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type the stream represents.

Remarks

The owner is resolved from the first event as in CreateAsync<TAggregate>(Guid, object, CancellationToken), and the events after it take the same owner.

Exceptions

SessionRequiredException

No session context is set on the current scope.

ExistsAsync(Guid, CancellationToken)

Returns true if the stream exists in the event store.

Task<bool> ExistsAsync(Guid streamId, CancellationToken cancellationToken = default)

Parameters

streamId Guid

cancellationToken CancellationToken

Returns

Task<bool>

GetCurrentVersionAsync(Guid, CancellationToken)

Returns the head version of the stream, or 0 if it does not exist.

Task<long> GetCurrentVersionAsync(Guid streamId, CancellationToken cancellationToken = default)

Parameters

streamId Guid

cancellationToken CancellationToken

Returns

Task<long>

SaveChangesAsync(CancellationToken)

Flush every pending append/create to the underlying write store. Throws ConcurrencyException if another writer committed first.

Task SaveChangesAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Honoured while the changes are written; a commit once begun runs to its end on the framework's stores, so a save the store committed is never reported as cancelled. A handover after the commit that the cancellation stops before the bundle is published or recorded surfaces as CommittedEventsNotPublishedException.

Returns

Task

Remarks

A save clears what was staged whether it succeeds or fails. After a failure, append the events again before saving again, as after a ConcurrencyException: a save with nothing staged writes and publishes nothing. The one exception is CommittedEventsNotPublishedException: its events are already recorded, and appending them again would record them twice.

Exceptions

OperationCanceledException

The save was cancelled before it committed; no events were written.

ConcurrencyException

Another writer beat this one to the stream's head version.

CommittedEventsNotPublishedException

The events were committed, but handing their bundle on failed afterwards. Do not append them again.

SessionRequiredException

No session context is set on the current scope.