Filter observers by event source and stream

Use [FromEventSource<TSource>("Stream")] on a reactor or a reducer when it should handle only the events appended to one stream of a definition. The filter replaces the pair of strings in [EventSourceType] and [EventStreamType].

Filter a reactor

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSources;
using Cratis.Chronicle.Reactors;

[EventType]
public record MoneyDeposited(decimal Amount);

[EventSource("Account")]
[EventStream("Onboarding")]
[EventStream("Transactions")]
public class AccountEventSource : IEventSource;

// Receives MoneyDeposited only when it was appended to the Transactions stream of the Account event source.
[FromEventSource<AccountEventSource>("Transactions")]
public class StatementReactor : IReactor
{
    public Task Deposited(MoneyDeposited @event, EventContext context) => Task.CompletedTask;
}

View C# snippet source on GitHub

The reactor receives MoneyDeposited only when it was appended to the Transactions stream of the Account event source. A deposit appended through another definition, or to the Onboarding stream, is skipped.

Filter a reducer

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSources;
using Cratis.Chronicle.Reducers;

public record DepositTotal(decimal Amount);

[FromEventSource<AccountEventSource>("Transactions")]
public class DepositTotalReducer : IReducerFor<DepositTotal>
{
    public DepositTotal Deposited(MoneyDeposited @event, DepositTotal? current, EventContext context) =>
        new((current?.Amount ?? 0m) + @event.Amount);
}

View C# snippet source on GitHub

How the match works

The attribute resolves to two values when the observer is registered: the definition’s name as the event source type, and the stream name as the event stream type. An event passes when its stored event source type and event stream type equal them. Chronicle compares the stored values, not the definition type, so an event appended with the same strings without a definition also matches.

If the definition or the stream cannot be resolved when the observer registers, the client throws UnknownEventSource or EventStreamDoesNotBelongToEventSource.

You can combine the typed filter with [FilterEventsByTag]. All configured categories must match. Filters apply to reactors and reducers. A projection ignores them, which CHR0041 reports.

Build-time checks

The code analyzer checks typed filters while you compile:

Rule	Severity	Reports
CHR0055	Error	The stream is not declared by the definition
CHR0056	Error	[EventSourceType] or [EventStreamType] on the same type contradicts the typed filter

An explicit attribute that repeats the same value as the typed filter is allowed, but it adds nothing.

Filtering by string

[EventSourceType] and [EventStreamType] still work without a definition. See Filter by event source type and Filter by event stream type. To move from the strings to the typed form, read Migrate from string routing.

Next steps

• Scenarios: a reactor that follows one process stream.
• Dynamic Consistency Boundary: the decision happens on the write side, and observers react afterwards.
• Reactors
• Dynamic Consistency Boundary in Chronicle: what consistency covers and what observers do not.