Filter reducers by appended event metadata

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

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 ReducersFilteringOrderPlaced(decimal TotalAmount);

public class ReducersFilteringMetadataExampleService(IEventLog eventLog)
{
    public Task PlaceOrder(decimal totalAmount) =>
        eventLog.Append(
            EventSourceId.New(),
            new ReducersFilteringOrderPlaced(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 ReducersFilteringOrderPlaced(val totalAmount: Double)

class ReducersFilteringMetadataExampleService(private val eventLog: IEventLog) {
    suspend fun placeOrder(eventSourceId: String, totalAmount: Double) =
        eventLog.append(
            eventSourceId,
            ReducersFilteringOrderPlaced(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 ReducersFilteringOrderPlaced(double totalAmount) {}

class ReducersFilteringMetadataExampleService {
    private final IEventLog eventLog;

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

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

Elixir

defmodule ReducersFilteringOrderPlaced do
  use Chronicle.Events.EventType, id: "reducers-filtering-order-placed"

  defstruct [:total_amount]
end

defmodule ReducersFilteringMetadataExampleService do
  alias ReducersFilteringOrderPlaced

  def place_order(order_id, total_amount) do
    Chronicle.append(
      order_id,
      %ReducersFilteringOrderPlaced{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 reducer class. Elixir’s use Chronicle.Reducers.Reducer and TypeScript’s reducer() register a reducer without a way to attach these filters.

Filter by tag

Append an event with a tag and place [FilterEventsByTag] on the reducer to accumulate state only from tagged events.

C#

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

[EventType]
public record ReducersFilteringByTagOrderPlaced(decimal TotalAmount);

public record ReducersFilteringPriorityOrderTotals(int Count, decimal Total);

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

[FilterEventsByTag("priority")]
public class ReducersFilteringPriorityOrderTotalsReducer : IReducerFor<ReducersFilteringPriorityOrderTotals>
{
    public ReducersFilteringPriorityOrderTotals Placed(ReducersFilteringByTagOrderPlaced @event, ReducersFilteringPriorityOrderTotals? current, EventContext context) =>
        new((current?.Count ?? 0) + 1, (current?.Total ?? 0m) + @event.TotalAmount);
}

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.Reducer
import io.cratis.chronicle.readModels.ReadModel

@EventType
data class ReducersFilteringByTagOrderPlaced(val totalAmount: Double)

@ReadModel
data class ReducersFilteringPriorityOrderTotals(val count: Int = 0, val total: Double = 0.0)

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

@Reducer
@FilterEventsByTag("priority")
class ReducersFilteringPriorityOrderTotalsReducer {
    fun placed(event: ReducersFilteringByTagOrderPlaced, current: ReducersFilteringPriorityOrderTotals?, context: EventContext) =
        ReducersFilteringPriorityOrderTotals((current?.count ?: 0) + 1, (current?.total ?: 0.0) + event.totalAmount)
}

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.Reducer;
import io.cratis.chronicle.readModels.ReadModel;

@EventType
record ReducersFilteringByTagOrderPlaced(double totalAmount) {}

@ReadModel
record ReducersFilteringPriorityOrderTotals(int count, double total) {
    ReducersFilteringPriorityOrderTotals() {
        this(0, 0.0);
    }
}

class ReducersFilteringByTagOrderService {
    private final IEventLog eventLog;

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

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

@Reducer
@FilterEventsByTag("priority")
class ReducersFilteringPriorityOrderTotalsReducer {
    ReducersFilteringPriorityOrderTotals placed(ReducersFilteringByTagOrderPlaced event, ReducersFilteringPriorityOrderTotals current, EventContext context) {
        int count = current == null ? 0 : current.count();
        double total = current == null ? 0.0 : current.total();
        return new ReducersFilteringPriorityOrderTotals(count + 1, total + event.totalAmount());
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

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

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

class ReducersFilteringPriorityOrderTotals {
    count = 0;
    total = 0;
}

@reducer('', undefined, ReducersFilteringPriorityOrderTotals)
@filterEventsByTag('priority')
class ReducersFilteringPriorityOrderTotalsReducer {
    reducersFilteringByTagOrderPlaced(
        event: ReducersFilteringByTagOrderPlaced,
        current: ReducersFilteringPriorityOrderTotals | undefined
    ): ReducersFilteringPriorityOrderTotals {
        return {
            count: (current?.count ?? 0) + 1,
            total: (current?.total ?? 0) + event.totalAmount
        };
    }
}

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

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

PriorityOrderTotalsReducer updates only when the appended event carries the priority tag. Orders without that tag do not affect this read model.

Multiple filter tags

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

C#

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

[EventType]
public record ReducersFilteringMultiTagOrderPlaced(decimal TotalAmount);

public record ReducersFilteringFastTrackOrderTotals(int Count);

[FilterEventsByTag("priority")]
[FilterEventsByTag("express")]
public class ReducersFilteringFastTrackOrderTotalsReducer : IReducerFor<ReducersFilteringFastTrackOrderTotals>
{
    public ReducersFilteringFastTrackOrderTotals Placed(ReducersFilteringMultiTagOrderPlaced @event, ReducersFilteringFastTrackOrderTotals? current, EventContext context) =>
        new((current?.Count ?? 0) + 1);
}

Kotlin

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.observation.FilterEventsByTag
import io.cratis.chronicle.observation.Reducer
import io.cratis.chronicle.readModels.ReadModel

@EventType
data class ReducersFilteringMultiTagOrderPlaced(val totalAmount: Double)

@ReadModel
data class ReducersFilteringFastTrackOrderTotals(val count: Int = 0)

@Reducer
@FilterEventsByTag("priority")
@FilterEventsByTag("express")
class ReducersFilteringFastTrackOrderTotalsReducer {
    fun placed(event: ReducersFilteringMultiTagOrderPlaced, current: ReducersFilteringFastTrackOrderTotals?, context: EventContext) =
        ReducersFilteringFastTrackOrderTotals((current?.count ?: 0) + 1)
}

Java

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.observation.FilterEventsByTag;
import io.cratis.chronicle.observation.Reducer;
import io.cratis.chronicle.readModels.ReadModel;

@EventType
record ReducersFilteringMultiTagOrderPlaced(double totalAmount) {}

@ReadModel
record ReducersFilteringFastTrackOrderTotals(int count) {
    ReducersFilteringFastTrackOrderTotals() {
        this(0);
    }
}

@Reducer
@FilterEventsByTag("priority")
@FilterEventsByTag("express")
class ReducersFilteringFastTrackOrderTotalsReducer {
    ReducersFilteringFastTrackOrderTotals placed(ReducersFilteringMultiTagOrderPlaced event, ReducersFilteringFastTrackOrderTotals current, EventContext context) {
        int count = current == null ? 0 : current.count();
        return new ReducersFilteringFastTrackOrderTotals(count + 1);
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

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

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

class ReducersFilteringFastTrackOrderTotals {
    count = 0;
}

@reducer('', undefined, ReducersFilteringFastTrackOrderTotals)
@filterEventsByTag('priority')
@filterEventsByTag('express')
class ReducersFilteringFastTrackOrderTotalsReducer {
    reducersFilteringMultiTagOrderPlaced(
        _event: ReducersFilteringMultiTagOrderPlaced,
        current: ReducersFilteringFastTrackOrderTotals | undefined
    ): ReducersFilteringFastTrackOrderTotals {
        return { count: (current?.count ?? 0) + 1 };
    }
}

Filter by event source type

Place [EventSourceType] on the reducer to accumulate state only from events appended with a matching eventSourceType:

C#

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

[EventType]
public record ReducersFilteringInvoiceIssued(decimal Amount);

public record ReducersFilteringCustomerInvoiceTotal(decimal Amount);

public class ReducersFilteringInvoicingService(IEventLog eventLog)
{
    public Task IssueCustomerInvoice(decimal amount) =>
        eventLog.Append(
            EventSourceId.New(),
            new ReducersFilteringInvoiceIssued(amount),
            eventSourceType: "customer");
}

[EventSourceType("customer")]
public class ReducersFilteringCustomerInvoiceTotalReducer : IReducerFor<ReducersFilteringCustomerInvoiceTotal>
{
    public ReducersFilteringCustomerInvoiceTotal Issued(ReducersFilteringInvoiceIssued @event, ReducersFilteringCustomerInvoiceTotal? current, EventContext context) =>
        new((current?.Amount ?? 0m) + @event.Amount);
}

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.Reducer
import io.cratis.chronicle.readModels.ReadModel

@EventType
data class ReducersFilteringInvoiceIssued(val amount: Double)

@ReadModel
data class ReducersFilteringCustomerInvoiceTotal(val amount: Double = 0.0)

class ReducersFilteringInvoicingService(private val eventLog: IEventLog) {
    suspend fun issueCustomerInvoice(eventSourceId: String, amount: Double) =
        eventLog.append(
            eventSourceId,
            ReducersFilteringInvoiceIssued(amount),
            AppendOptions(eventSourceType = "customer")
        )
}

@Reducer
@EventSourceType("customer")
class ReducersFilteringCustomerInvoiceTotalReducer {
    fun issued(event: ReducersFilteringInvoiceIssued, current: ReducersFilteringCustomerInvoiceTotal?, context: EventContext) =
        ReducersFilteringCustomerInvoiceTotal((current?.amount ?: 0.0) + event.amount)
}

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.Reducer;
import io.cratis.chronicle.readModels.ReadModel;

@EventType
record ReducersFilteringInvoiceIssued(double amount) {}

@ReadModel
record ReducersFilteringCustomerInvoiceTotal(double amount) {
    ReducersFilteringCustomerInvoiceTotal() {
        this(0.0);
    }
}

class ReducersFilteringInvoicingService {
    private final IEventLog eventLog;

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

    void issueCustomerInvoice(String eventSourceId, double amount) {
        EventLogJavaBridge.append(
            eventLog,
            eventSourceId,
            new ReducersFilteringInvoiceIssued(amount),
            new AppendOptionsBuilder().eventSourceType("customer").build());
    }
}

@Reducer
@EventSourceType("customer")
class ReducersFilteringCustomerInvoiceTotalReducer {
    ReducersFilteringCustomerInvoiceTotal issued(ReducersFilteringInvoiceIssued event, ReducersFilteringCustomerInvoiceTotal current, EventContext context) {
        double amount = current == null ? 0.0 : current.amount();
        return new ReducersFilteringCustomerInvoiceTotal(amount + event.amount());
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

If the same event is appended with eventSourceType: "supplier", this reducer does not receive it.

Filter by event stream type

Place [EventStreamType] on the reducer to accumulate state only from events appended to a matching stream type:

C#

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

[EventType]
public record ReducersFilteringShipmentSent(decimal ShippingCost);

public record ReducersFilteringShippingTotals(int Count, decimal TotalCost);

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

[EventStreamType("shipping")]
public class ReducersFilteringShippingTotalsReducer : IReducerFor<ReducersFilteringShippingTotals>
{
    public ReducersFilteringShippingTotals Sent(ReducersFilteringShipmentSent @event, ReducersFilteringShippingTotals? current, EventContext context) =>
        new((current?.Count ?? 0) + 1, (current?.TotalCost ?? 0m) + @event.ShippingCost);
}

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.Reducer
import io.cratis.chronicle.readModels.ReadModel

@EventType
data class ReducersFilteringShipmentSent(val shippingCost: Double)

@ReadModel
data class ReducersFilteringShippingTotals(val count: Int = 0, val totalCost: Double = 0.0)

class ReducersFilteringShippingService(private val eventLog: IEventLog) {
    suspend fun send(eventSourceId: String, shippingCost: Double) =
        eventLog.append(
            eventSourceId,
            ReducersFilteringShipmentSent(shippingCost),
            AppendOptions(eventStreamType = "shipping")
        )
}

@Reducer
@EventStreamType("shipping")
class ReducersFilteringShippingTotalsReducer {
    fun sent(event: ReducersFilteringShipmentSent, current: ReducersFilteringShippingTotals?, context: EventContext) =
        ReducersFilteringShippingTotals((current?.count ?: 0) + 1, (current?.totalCost ?: 0.0) + event.shippingCost)
}

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.Reducer;
import io.cratis.chronicle.readModels.ReadModel;

@EventType
record ReducersFilteringShipmentSent(double shippingCost) {}

@ReadModel
record ReducersFilteringShippingTotals(int count, double totalCost) {
    ReducersFilteringShippingTotals() {
        this(0, 0.0);
    }
}

class ReducersFilteringShippingService {
    private final IEventLog eventLog;

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

    void send(String eventSourceId, double shippingCost) {
        EventLogJavaBridge.append(
            eventLog,
            eventSourceId,
            new ReducersFilteringShipmentSent(shippingCost),
            new AppendOptionsBuilder().eventStreamType("shipping").build());
    }
}

@Reducer
@EventStreamType("shipping")
class ReducersFilteringShippingTotalsReducer {
    ReducersFilteringShippingTotals sent(ReducersFilteringShipmentSent event, ReducersFilteringShippingTotals current, EventContext context) {
        int count = current == null ? 0 : current.count();
        double totalCost = current == null ? 0.0 : current.totalCost();
        return new ReducersFilteringShippingTotals(count + 1, totalCost + event.shippingCost());
    }
}

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 reducer. Chronicle requires all filter categories to match:

C#

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

[EventType]
public record ReducersFilteringCombineOrderPlaced(decimal TotalAmount);

public record ReducersFilteringPremiumFulfillmentTotals(int Count, decimal Total);

public class ReducersFilteringCombineOrderService(IEventLog eventLog)
{
    public Task PlacePremiumOrder(decimal totalAmount) =>
        eventLog.Append(
            EventSourceId.New(),
            new ReducersFilteringCombineOrderPlaced(totalAmount),
            tags: ["premium"],
            eventSourceType: "order",
            eventStreamType: "fulfillment");
}

[FilterEventsByTag("premium")]
[EventSourceType("order")]
[EventStreamType("fulfillment")]
public class ReducersFilteringPremiumFulfillmentTotalsReducer : IReducerFor<ReducersFilteringPremiumFulfillmentTotals>
{
    public ReducersFilteringPremiumFulfillmentTotals Placed(ReducersFilteringCombineOrderPlaced @event, ReducersFilteringPremiumFulfillmentTotals? current, EventContext context) =>
        new((current?.Count ?? 0) + 1, (current?.Total ?? 0m) + @event.TotalAmount);
}

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.EventStreamType
import io.cratis.chronicle.observation.FilterEventsByTag
import io.cratis.chronicle.observation.Reducer
import io.cratis.chronicle.readModels.ReadModel

@EventType
data class ReducersFilteringCombineOrderPlaced(val totalAmount: Double)

@ReadModel
data class ReducersFilteringPremiumFulfillmentTotals(val count: Int = 0, val total: Double = 0.0)

class ReducersFilteringCombineOrderService(private val eventLog: IEventLog) {
    suspend fun placePremiumOrder(eventSourceId: String, totalAmount: Double) =
        eventLog.append(
            eventSourceId,
            ReducersFilteringCombineOrderPlaced(totalAmount),
            AppendOptions(tags = listOf("premium"), eventSourceType = "order", eventStreamType = "fulfillment")
        )
}

@Reducer
@FilterEventsByTag("premium")
@EventSourceType("order")
@EventStreamType("fulfillment")
class ReducersFilteringPremiumFulfillmentTotalsReducer {
    fun placed(event: ReducersFilteringCombineOrderPlaced, current: ReducersFilteringPremiumFulfillmentTotals?, context: EventContext) =
        ReducersFilteringPremiumFulfillmentTotals((current?.count ?: 0) + 1, (current?.total ?: 0.0) + event.totalAmount)
}

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.EventStreamType;
import io.cratis.chronicle.observation.FilterEventsByTag;
import io.cratis.chronicle.observation.Reducer;
import io.cratis.chronicle.readModels.ReadModel;

@EventType
record ReducersFilteringCombineOrderPlaced(double totalAmount) {}

@ReadModel
record ReducersFilteringPremiumFulfillmentTotals(int count, double total) {
    ReducersFilteringPremiumFulfillmentTotals() {
        this(0, 0.0);
    }
}

class ReducersFilteringCombineOrderService {
    private final IEventLog eventLog;

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

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

@Reducer
@FilterEventsByTag("premium")
@EventSourceType("order")
@EventStreamType("fulfillment")
class ReducersFilteringPremiumFulfillmentTotalsReducer {
    ReducersFilteringPremiumFulfillmentTotals placed(ReducersFilteringCombineOrderPlaced event, ReducersFilteringPremiumFulfillmentTotals current, EventContext context) {
        int count = current == null ? 0 : current.count();
        double total = current == null ? 0.0 : current.total();
        return new ReducersFilteringPremiumFulfillmentTotals(count + 1, total + event.totalAmount());
    }
}

Elixir

Elixir does not support this workflow yet.

TypeScript

TypeScript does not support this workflow yet.

The reducer only receives events that match all three: the premium tag, the order event source type, and the fulfillment stream type.

Tagging the reducer vs filtering events

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

C#

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

[EventType]
public record ReducersFilteringTagVsFilterOrderPlaced(decimal TotalAmount);

public record ReducersFilteringTagVsFilterTotals(int Count, decimal Total);

// These labels appear on the reducer definition — they do not affect dispatch
[Tag("reporting")]
[Tag("premium")]
public class ReducersFilteringLabeledFulfillmentTotalsReducer : IReducerFor<ReducersFilteringTagVsFilterTotals>
{
    public ReducersFilteringTagVsFilterTotals Placed(ReducersFilteringTagVsFilterOrderPlaced @event, ReducersFilteringTagVsFilterTotals? current, EventContext context) =>
        new((current?.Count ?? 0) + 1, (current?.Total ?? 0m) + @event.TotalAmount);
}

// These filter which events are dispatched to the reducer
[FilterEventsByTag("premium")]
[EventSourceType("order")]
public class ReducersFilteringFilteredFulfillmentTotalsReducer : IReducerFor<ReducersFilteringTagVsFilterTotals>
{
    public ReducersFilteringTagVsFilterTotals Placed(ReducersFilteringTagVsFilterOrderPlaced @event, ReducersFilteringTagVsFilterTotals? current, EventContext context) =>
        new((current?.Count ?? 0) + 1, (current?.Total ?? 0m) + @event.TotalAmount);
}

Kotlin

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.observation.EventSourceType
import io.cratis.chronicle.observation.FilterEventsByTag
import io.cratis.chronicle.observation.Reducer
import io.cratis.chronicle.observation.Tag
import io.cratis.chronicle.readModels.ReadModel

@EventType
data class ReducersFilteringTagVsFilterOrderPlaced(val totalAmount: Double)

@ReadModel
data class ReducersFilteringTagVsFilterTotals(val count: Int = 0, val total: Double = 0.0)

// These labels appear on the reducer definition - they do not affect dispatch
@Reducer
@Tag("reporting")
@Tag("premium")
class ReducersFilteringLabeledFulfillmentTotalsReducer {
    fun placed(event: ReducersFilteringTagVsFilterOrderPlaced, current: ReducersFilteringTagVsFilterTotals?, context: EventContext) =
        ReducersFilteringTagVsFilterTotals((current?.count ?: 0) + 1, (current?.total ?: 0.0) + event.totalAmount)
}

// These filter which events are dispatched to the reducer
@Reducer
@FilterEventsByTag("premium")
@EventSourceType("order")
class ReducersFilteringFilteredFulfillmentTotalsReducer {
    fun placed(event: ReducersFilteringTagVsFilterOrderPlaced, current: ReducersFilteringTagVsFilterTotals?, context: EventContext) =
        ReducersFilteringTagVsFilterTotals((current?.count ?: 0) + 1, (current?.total ?: 0.0) + event.totalAmount)
}

Java

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.observation.EventSourceType;
import io.cratis.chronicle.observation.FilterEventsByTag;
import io.cratis.chronicle.observation.Reducer;
import io.cratis.chronicle.observation.Tag;
import io.cratis.chronicle.readModels.ReadModel;

@EventType
record ReducersFilteringTagVsFilterOrderPlaced(double totalAmount) {}

@ReadModel
record ReducersFilteringTagVsFilterTotals(int count, double total) {
    ReducersFilteringTagVsFilterTotals() {
        this(0, 0.0);
    }
}

// These labels appear on the reducer definition - they do not affect dispatch
@Reducer
@Tag("reporting")
@Tag("premium")
class ReducersFilteringLabeledFulfillmentTotalsReducer {
    ReducersFilteringTagVsFilterTotals placed(ReducersFilteringTagVsFilterOrderPlaced event, ReducersFilteringTagVsFilterTotals current, EventContext context) {
        int count = current == null ? 0 : current.count();
        double total = current == null ? 0.0 : current.total();
        return new ReducersFilteringTagVsFilterTotals(count + 1, total + event.totalAmount());
    }
}

// These filter which events are dispatched to the reducer
@Reducer
@FilterEventsByTag("premium")
@EventSourceType("order")
class ReducersFilteringFilteredFulfillmentTotalsReducer {
    ReducersFilteringTagVsFilterTotals placed(ReducersFilteringTagVsFilterOrderPlaced event, ReducersFilteringTagVsFilterTotals current, EventContext context) {
        int count = current == null ? 0 : current.count();
        double total = current == null ? 0.0 : current.total();
        return new ReducersFilteringTagVsFilterTotals(count + 1, total + event.totalAmount());
    }
}

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