Filter reducers and reactors by event stream type

Use [EventStreamType] on a reducer or reactor when it should only handle events appended to a specific stream type.

How the match works

Chronicle compares the observer attribute to the eventStreamType: value used when appending the event. If the values do not match, the reducer or reactor is skipped.

Filter a reactor by event stream type

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Reactors;

[EventType]
public record FilterByStreamTypePaymentCaptured(decimal Amount);

public class FilterByStreamTypePaymentsService(IEventLog eventLog)
{
    public Task Capture(decimal amount) =>
        eventLog.Append(
            EventSourceId.New(),
            new FilterByStreamTypePaymentCaptured(amount),
            eventStreamType: "payments");
}

[EventStreamType("payments")]
public class FilterByStreamTypePaymentNotificationsReactor : IReactor
{
    public Task Captured(FilterByStreamTypePaymentCaptured @event, EventContext context) => Task.CompletedTask;
}

Kotlin

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.eventSequences.AppendOptions
import io.cratis.chronicle.observation.EventStreamType
import io.cratis.chronicle.observation.Reactor

@EventType
data class FilterByStreamTypePaymentCaptured(val amount: Double = 0.0)

suspend fun capture(store: IEventStore, eventSourceId: String, amount: Double) =
    store.eventLog.append(eventSourceId, FilterByStreamTypePaymentCaptured(amount), AppendOptions(eventStreamType = "payments"))

@EventStreamType("payments")
@Reactor
class FilterByStreamTypePaymentNotificationsReactor {
    fun paymentCaptured(event: FilterByStreamTypePaymentCaptured, context: EventContext) {
        // Only handles events appended to the "payments" stream type
    }
}

Java

import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.AppendOptions;
import io.cratis.chronicle.eventSequences.AppendResult;
import io.cratis.chronicle.observation.EventStreamType;
import io.cratis.chronicle.observation.Reactor;

import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;

@EventType
record FilterByStreamTypePaymentCaptured(double amount) {}

@EventStreamType("payments")
@Reactor
class FilterByStreamTypePaymentNotificationsReactor {
    public void paymentCaptured(FilterByStreamTypePaymentCaptured event, EventContext context) {
        // Only handles events appended to the "payments" stream type
    }
}

class EventsFilteringByEventStreamTypeReactor {
    AppendResult capture(EventStore store, String eventSourceId, double amount) {
        AppendOptions options = new AppendOptionsBuilder().eventStreamType("payments").build();
        return EventLogJavaBridge.append(store.getEventLog(), eventSourceId, new FilterByStreamTypePaymentCaptured(amount), options);
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

The reactor only handles events appended to the payments stream type.

Filter a reducer by event stream type

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Reducers;

[EventType]
public record FilterByStreamTypeShipmentSent(decimal ShippingCost);

public record FilterByStreamTypeShippingTotals(decimal ShippingCost);

[EventStreamType("shipping")]
public class FilterByStreamTypeShippingTotalsReducer : IReducerFor<FilterByStreamTypeShippingTotals>
{
    public FilterByStreamTypeShippingTotals Sent(FilterByStreamTypeShipmentSent @event, FilterByStreamTypeShippingTotals? current, EventContext context) =>
        new((current?.ShippingCost ?? 0m) + @event.ShippingCost);
}

public class FilterByStreamTypeShippingService(IEventLog eventLog)
{
    public Task Send(decimal shippingCost) =>
        eventLog.Append(
            EventSourceId.New(),
            new FilterByStreamTypeShipmentSent(shippingCost),
            eventStreamType: "shipping");
}

Kotlin

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.eventSequences.AppendOptions
import io.cratis.chronicle.observation.EventStreamType
import io.cratis.chronicle.observation.Reducer

@EventType
data class FilterByStreamTypeShipmentSent(val shippingCost: Double = 0.0)

data class FilterByStreamTypeShippingTotals(val shippingCost: Double = 0.0)

@EventStreamType("shipping")
@Reducer
class FilterByStreamTypeShippingTotalsReducer {
    fun shipmentSent(event: FilterByStreamTypeShipmentSent, current: FilterByStreamTypeShippingTotals?): FilterByStreamTypeShippingTotals =
        FilterByStreamTypeShippingTotals((current?.shippingCost ?: 0.0) + event.shippingCost)
}

suspend fun send(store: IEventStore, eventSourceId: String, shippingCost: Double) =
    store.eventLog.append(eventSourceId, FilterByStreamTypeShipmentSent(shippingCost), AppendOptions(eventStreamType = "shipping"))

Java

import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.AppendOptions;
import io.cratis.chronicle.eventSequences.AppendResult;
import io.cratis.chronicle.observation.EventStreamType;
import io.cratis.chronicle.observation.Reducer;

import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;

@EventType
record FilterByStreamTypeShipmentSent(double shippingCost) {}

record FilterByStreamTypeShippingTotals(double shippingCost) {}

@EventStreamType("shipping")
@Reducer
class FilterByStreamTypeShippingTotalsReducer {
    public FilterByStreamTypeShippingTotals shipmentSent(FilterByStreamTypeShipmentSent event, FilterByStreamTypeShippingTotals current) {
        double currentCost = current != null ? current.shippingCost() : 0.0;
        return new FilterByStreamTypeShippingTotals(currentCost + event.shippingCost());
    }
}

class EventsFilteringByEventStreamTypeReducer {
    AppendResult send(EventStore store, String eventSourceId, double shippingCost) {
        AppendOptions options = new AppendOptionsBuilder().eventStreamType("shipping").build();
        return EventLogJavaBridge.append(store.getEventLog(), eventSourceId, new FilterByStreamTypeShipmentSent(shippingCost), options);
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

If the same event is appended with another stream type, such as eventStreamType: "returns", this reducer does not receive it.

When to choose stream type filtering

Use stream type filtering when the same event source can produce distinct processing flows and you want a reducer or reactor to observe only one of those flows.