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:

C#

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");
}

Kotlin

import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.eventSequences.AppendOptions
import io.cratis.chronicle.eventSequences.IEventLog

@EventType
data class ReactorsFilteringOrderPlaced(val totalAmount: Double)

class ReactorsFilteringMetadataExampleService(private val eventLog: IEventLog) {
    suspend fun placeOrder(eventSourceId: String, totalAmount: Double) =
        eventLog.append(
            eventSourceId,
            ReactorsFilteringOrderPlaced(totalAmount),
            AppendOptions(
                tags = listOf("priority"),
                eventSourceType = "order",
                eventStreamType = "fulfillment"
            )
        )
}

Java

import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.IEventLog;
import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;

@EventType
record ReactorsFilteringOrderPlaced(double totalAmount) {}

class ReactorsFilteringMetadataExampleService {
    private final IEventLog eventLog;

    ReactorsFilteringMetadataExampleService(IEventLog eventLog) {
        this.eventLog = eventLog;
    }

    void placeOrder(String eventSourceId, double totalAmount) {
        EventLogJavaBridge.append(
            eventLog,
            eventSourceId,
            new ReactorsFilteringOrderPlaced(totalAmount),
            new AppendOptionsBuilder()
                .tag("priority")
                .eventSourceType("order")
                .eventStreamType("fulfillment")
                .build());
    }
}

Elixir

defmodule ReactorsFilteringOrderPlaced do
  use Chronicle.Events.EventType, id: "reactors-filtering-order-placed"

  defstruct [:total_amount]
end

defmodule ReactorsFilteringMetadataExampleService do
  alias ReactorsFilteringOrderPlaced

  def place_order(order_id, total_amount) do
    Chronicle.append(
      order_id,
      %ReactorsFilteringOrderPlaced{total_amount: total_amount},
      tags: ["priority"],
      event_source_type: "order",
      event_stream_type: "fulfillment"
    )
  end
end

TypeScript

TypeScript does not support this workflow yet.

Client coverage

Kotlin and Java attach the same filters with @FilterEventsByTag, @EventSourceType, and @EventStreamType on the reactor class. Elixir’s use Chronicle.Reactors.Reactor and TypeScript’s reactor() register a reactor without a way to attach these filters.

Filter by tag

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

C#

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;
}

Kotlin

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.eventSequences.AppendOptions
import io.cratis.chronicle.eventSequences.IEventLog
import io.cratis.chronicle.observation.FilterEventsByTag
import io.cratis.chronicle.observation.Reactor

@EventType
data class ReactorsFilteringByTagOrderPlaced(val totalAmount: Double)

class ReactorsFilteringByTagOrderService(private val eventLog: IEventLog) {
    suspend fun placePriorityOrder(eventSourceId: String, totalAmount: Double) =
        eventLog.append(
            eventSourceId,
            ReactorsFilteringByTagOrderPlaced(totalAmount),
            AppendOptions(tags = listOf("priority"))
        )
}

@Reactor
@FilterEventsByTag("priority")
class ReactorsFilteringPriorityOrderNotifier {
    fun placed(event: ReactorsFilteringByTagOrderPlaced, context: EventContext) {
    }
}

Java

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.IEventLog;
import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;
import io.cratis.chronicle.observation.FilterEventsByTag;
import io.cratis.chronicle.observation.Reactor;

@EventType
record ReactorsFilteringByTagOrderPlaced(double totalAmount) {}

class ReactorsFilteringByTagOrderService {
    private final IEventLog eventLog;

    ReactorsFilteringByTagOrderService(IEventLog eventLog) {
        this.eventLog = eventLog;
    }

    void placePriorityOrder(String eventSourceId, double totalAmount) {
        EventLogJavaBridge.append(
            eventLog,
            eventSourceId,
            new ReactorsFilteringByTagOrderPlaced(totalAmount),
            new AppendOptionsBuilder().tag("priority").build());
    }
}

@Reactor
@FilterEventsByTag("priority")
class ReactorsFilteringPriorityOrderNotifier {
    void placed(ReactorsFilteringByTagOrderPlaced event, EventContext context) {
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

import { EventContext, eventType, filterEventsByTag, IEventStore, reactor } from '@cratis/chronicle';

@eventType()
class ReactorsFilteringByTagOrderPlaced {
    constructor(readonly totalAmount: number) {}
}

class ReactorsFilteringByTagOrderService {
    constructor(private readonly store: IEventStore) {}

    async placePriorityOrder(eventSourceId: string, totalAmount: number): Promise<void> {
        await this.store.eventLog.append(
            eventSourceId,
            new ReactorsFilteringByTagOrderPlaced(totalAmount),
            { tags: ['priority'] });
    }
}

@reactor()
@filterEventsByTag('priority')
class ReactorsFilteringPriorityOrderNotifier {
    async reactorsFilteringByTagOrderPlaced(_event: ReactorsFilteringByTagOrderPlaced, _context: EventContext): Promise<void> {}
}

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:

C#

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;
}

Kotlin

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
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.
    }
}

Java

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
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.
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

import { EventContext, eventType, filterEventsByTag, reactor } from '@cratis/chronicle';

@eventType()
class ReactorsFilteringMultiTagOrderPlaced {
    constructor(readonly totalAmount: number) {}
}

@reactor()
@filterEventsByTag('priority')
@filterEventsByTag('express')
class ReactorsFilteringFastTrackOrderNotifier {
    async reactorsFilteringMultiTagOrderPlaced(_event: ReactorsFilteringMultiTagOrderPlaced, _context: EventContext): Promise<void> {}
}

Filter by event source type

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

C#

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;
}

Kotlin

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.eventSequences.AppendOptions
import io.cratis.chronicle.eventSequences.IEventLog
import io.cratis.chronicle.observation.EventSourceType
import io.cratis.chronicle.observation.Reactor

@EventType
data class ReactorsFilteringCustomerRegistered(val emailAddress: String)

class ReactorsFilteringCustomerService(private val eventLog: IEventLog) {
    suspend fun register(eventSourceId: String, emailAddress: String) =
        eventLog.append(
            eventSourceId,
            ReactorsFilteringCustomerRegistered(emailAddress),
            AppendOptions(eventSourceType = "customer")
        )
}

@Reactor
@EventSourceType("customer")
class ReactorsFilteringCustomerWelcomeReactor {
    fun registered(event: ReactorsFilteringCustomerRegistered, context: EventContext) {
    }
}

Java

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.IEventLog;
import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;
import io.cratis.chronicle.observation.EventSourceType;
import io.cratis.chronicle.observation.Reactor;

@EventType
record ReactorsFilteringCustomerRegistered(String emailAddress) {}

class ReactorsFilteringCustomerService {
    private final IEventLog eventLog;

    ReactorsFilteringCustomerService(IEventLog eventLog) {
        this.eventLog = eventLog;
    }

    void register(String eventSourceId, String emailAddress) {
        EventLogJavaBridge.append(
            eventLog,
            eventSourceId,
            new ReactorsFilteringCustomerRegistered(emailAddress),
            new AppendOptionsBuilder().eventSourceType("customer").build());
    }
}

@Reactor
@EventSourceType("customer")
class ReactorsFilteringCustomerWelcomeReactor {
    void registered(ReactorsFilteringCustomerRegistered event, EventContext context) {
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

Filter by event stream type

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

C#

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;
}

Kotlin

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

@EventType
data class ReactorsFilteringPaymentCaptured(val amount: Double)

class ReactorsFilteringPaymentsService(private val eventLog: IEventLog) {
    suspend fun capture(eventSourceId: String, amount: Double) =
        eventLog.append(
            eventSourceId,
            ReactorsFilteringPaymentCaptured(amount),
            AppendOptions(eventStreamType = "payments")
        )
}

@Reactor
@EventStreamType("payments")
class ReactorsFilteringPaymentReceivedNotifier {
    fun captured(event: ReactorsFilteringPaymentCaptured, context: EventContext) {
    }
}

Java

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.IEventLog;
import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;
import io.cratis.chronicle.observation.EventStreamType;
import io.cratis.chronicle.observation.Reactor;

@EventType
record ReactorsFilteringPaymentCaptured(double amount) {}

class ReactorsFilteringPaymentsService {
    private final IEventLog eventLog;

    ReactorsFilteringPaymentsService(IEventLog eventLog) {
        this.eventLog = eventLog;
    }

    void capture(String eventSourceId, double amount) {
        EventLogJavaBridge.append(
            eventLog,
            eventSourceId,
            new ReactorsFilteringPaymentCaptured(amount),
            new AppendOptionsBuilder().eventStreamType("payments").build());
    }
}

@Reactor
@EventStreamType("payments")
class ReactorsFilteringPaymentReceivedNotifier {
    void captured(ReactorsFilteringPaymentCaptured event, EventContext context) {
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

Combine multiple filters

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

C#

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;
}

Kotlin

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
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.
    }
}

Java

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
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.
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

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:

C#

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;
}

Kotlin

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
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
    ) {
    }
}

Java

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
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) {
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

Detailed guides

• Filter by tag
• Filter by event source type
• Filter by event stream type