Filter reactors by appended event metadata

A reactor observes all events that match its handler method signatures by default. You can restrict which events a reactor receives by placing filter attributes on the reactor class. Chronicle evaluates these filters before dispatching an event — events that do not match are dropped before they ever reach the reactor.

Filter attributes

Attribute	Filters by	Matches when
`[FilterEventsByTag("...")]`	Appended or static event tag	Any filter tag matches any tag on the appended event
`[EventSourceType("...")]`	Event source type set at append time	The event source type matches exactly
`[EventStreamType("...")]`	Event stream type set at append time	The event stream type matches exactly

These attributes correlate directly to the metadata you provide when appending an event:

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

[EventType]
public record ReactorsFilteringOrderPlaced(decimal TotalAmount);

public class ReactorsFilteringMetadataExampleService(IEventLog eventLog)
{
    public Task PlaceOrder(decimal totalAmount) =>
        eventLog.Append(
            EventSourceId.New(),
            new ReactorsFilteringOrderPlaced(totalAmount),
            tags: ["priority"],
            eventSourceType: "order",
            eventStreamType: "fulfillment");
}

Filter by tag

Append an event with a tag and place [FilterEventsByTag] on the reactor to receive only tagged events.

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

[EventType]
public record ReactorsFilteringByTagOrderPlaced(decimal TotalAmount);

public class ReactorsFilteringByTagOrderService(IEventLog eventLog)
{
    public Task PlacePriorityOrder(decimal totalAmount) =>
        eventLog.Append(
            EventSourceId.New(),
            new ReactorsFilteringByTagOrderPlaced(totalAmount),
            tags: ["priority"]);
}

[FilterEventsByTag("priority")]
public class ReactorsFilteringPriorityOrderNotifier : IReactor
{
    public Task Placed(ReactorsFilteringByTagOrderPlaced @event, EventContext context) =>
        Task.CompletedTask;
}

PriorityOrderNotifier receives the event because the append call includes the priority tag. Orders appended without that tag are not dispatched.

Multiple filter tags

Multiple [FilterEventsByTag] attributes widen the match. The reactor receives the event if the appended event has any of the configured tags:

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

[EventType]
public record ReactorsFilteringMultiTagOrderPlaced(decimal TotalAmount);

[FilterEventsByTag("priority")]
[FilterEventsByTag("express")]
public class ReactorsFilteringFastTrackOrderNotifier : IReactor
{
    public Task Placed(ReactorsFilteringMultiTagOrderPlaced @event, EventContext context) =>
        Task.CompletedTask;
}

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.observation.FilterEventsByTag
import io.cratis.chronicle.observation.Reactor

@EventType(id = "reactors-filtering-multi-tag-order-placed")
data class ReactorsFilteringMultiTagOrderPlaced(val totalAmount: Double)

@Reactor
@FilterEventsByTag("priority")
@FilterEventsByTag("express")
class ReactorsFilteringFastTrackOrderNotifier {
    fun placed(
        event: ReactorsFilteringMultiTagOrderPlaced,
        context: EventContext
    ) {
        // Only events appended with both tags reach this handler.
    }
}

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.observation.FilterEventsByTag;
import io.cratis.chronicle.observation.Reactor;

@EventType(id = "reactors-filtering-multi-tag-order-placed")
record ReactorsFilteringMultiTagOrderPlaced(double totalAmount) {}

@Reactor
@FilterEventsByTag("priority")
@FilterEventsByTag("express")
class ReactorsFilteringFastTrackOrderNotifier {
    void placed(ReactorsFilteringMultiTagOrderPlaced event, EventContext context) {
        // Only events appended with both tags reach this handler.
    }
}

Filter by event source type

Place [EventSourceType] on the reactor to receive only events appended with a matching eventSourceType:

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

[EventType]
public record ReactorsFilteringCustomerRegistered(string EmailAddress);

public class ReactorsFilteringCustomerService(IEventLog eventLog)
{
    public Task Register(string emailAddress) =>
        eventLog.Append(
            EventSourceId.New(),
            new ReactorsFilteringCustomerRegistered(emailAddress),
            eventSourceType: "customer");
}

[EventSourceType("customer")]
public class ReactorsFilteringCustomerWelcomeReactor : IReactor
{
    public Task Registered(ReactorsFilteringCustomerRegistered @event, EventContext context) =>
        Task.CompletedTask;
}

Filter by event stream type

Place [EventStreamType] on the reactor to receive only events appended to a matching stream type:

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

[EventType]
public record ReactorsFilteringPaymentCaptured(decimal Amount);

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

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

Combine multiple filters

You can combine [FilterEventsByTag], [EventSourceType], and [EventStreamType] on the same reactor. Chronicle requires all filter categories to match:

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

[EventType]
public record ReactorsFilteringShipmentDispatched(string TrackingNumber);

public class ReactorsFilteringShippingService(IEventLog eventLog)
{
    public Task Dispatch(string trackingNumber) =>
        eventLog.Append(
            EventSourceId.New(),
            new ReactorsFilteringShipmentDispatched(trackingNumber),
            tags: ["express"],
            eventSourceType: "shipment",
            eventStreamType: "logistics");
}

[FilterEventsByTag("express")]
[EventSourceType("shipment")]
[EventStreamType("logistics")]
public class ReactorsFilteringExpressShipmentNotifier : IReactor
{
    public Task Dispatched(ReactorsFilteringShipmentDispatched @event, EventContext context) =>
        Task.CompletedTask;
}

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.observation.EventSourceType
import io.cratis.chronicle.observation.EventStreamType
import io.cratis.chronicle.observation.FilterEventsByTag
import io.cratis.chronicle.observation.Reactor

@EventType(id = "reactors-filtering-shipment-dispatched")
data class ReactorsFilteringShipmentDispatched(val trackingNumber: String)

@Reactor
@FilterEventsByTag("priority")
@EventSourceType("Order")
@EventStreamType("Fulfilment")
class ReactorsFilteringShipmentNotifier {
    fun dispatched(
        event: ReactorsFilteringShipmentDispatched,
        context: EventContext
    ) {
        // Every filter has to match: the tag, the event source type, and the stream type.
    }
}

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.observation.EventSourceType;
import io.cratis.chronicle.observation.EventStreamType;
import io.cratis.chronicle.observation.FilterEventsByTag;
import io.cratis.chronicle.observation.Reactor;

@EventType(id = "reactors-filtering-shipment-dispatched")
record ReactorsFilteringShipmentDispatched(String trackingNumber) {}

@Reactor
@FilterEventsByTag("priority")
@EventSourceType("Order")
@EventStreamType("Fulfilment")
class ReactorsFilteringShipmentNotifier {
    void dispatched(ReactorsFilteringShipmentDispatched event, EventContext context) {
        // Every filter has to match: the tag, the event source type, and the stream type.
    }
}

The reactor only receives events that match all three: the express tag, the shipment event source type, and the logistics stream type.

Tagging the reactor vs filtering events

[Tag] and [Tags] on a reactor class label the reactor itself for organization and discoverability. They do not filter incoming events:

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

[EventType]
public record ReactorsFilteringTagVsFilterShipmentDispatched(string TrackingNumber);

// These labels appear on the reactor definition — they do not affect dispatch
[Tag("notifications")]
[Tag("express")]
public class ReactorsFilteringLabeledShipmentNotifier : IReactor
{
    public Task Dispatched(ReactorsFilteringTagVsFilterShipmentDispatched @event, EventContext context) =>
        Task.CompletedTask;
}

// These filter which events are dispatched to the reactor
[FilterEventsByTag("express")]
[EventSourceType("shipment")]
public class ReactorsFilteringFilteredShipmentNotifier : IReactor
{
    public Task Dispatched(ReactorsFilteringTagVsFilterShipmentDispatched @event, EventContext context) =>
        Task.CompletedTask;
}

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.observation.FilterEventsByTag
import io.cratis.chronicle.observation.Reactor
import io.cratis.chronicle.observation.Tag

@EventType(id = "reactors-filtering-tag-vs-filter-invoice-issued")
data class ReactorsFilteringInvoiceIssued(val amount: Double)

// @Tag labels the reactor itself and shows up in tooling - it changes nothing about
// what the reactor observes. @FilterEventsByTag is what narrows the event stream.
@Reactor
@Tag("finance", "owned-by-billing")
@FilterEventsByTag("audited")
class ReactorsFilteringInvoiceAuditor {
    fun issued(
        event: ReactorsFilteringInvoiceIssued,
        context: EventContext
    ) {
    }
}

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.observation.FilterEventsByTag;
import io.cratis.chronicle.observation.Reactor;
import io.cratis.chronicle.observation.Tag;

@EventType(id = "reactors-filtering-tag-vs-filter-invoice-issued")
record ReactorsFilteringInvoiceIssued(double amount) {}

// @Tag labels the reactor itself and shows up in tooling - it changes nothing about
// what the reactor observes. @FilterEventsByTag is what narrows the event stream.
@Reactor
@Tag({"finance", "owned-by-billing"})
@FilterEventsByTag("audited")
class ReactorsFilteringInvoiceAuditor {
    void issued(ReactorsFilteringInvoiceIssued event, EventContext context) {
    }
}