Getting Started with Reactors

Reactors observe events and run side effects. Use them for work such as sending notifications, calling an external API, publishing integration messages, or triggering a workflow in another system.

Reactors do not build queryable state. Use a projection or reducer when the outcome should be a read model.

Define the event

A reactor handles one or more event types. The event is still a normal Chronicle event: an immutable fact with a stable event type identity.

[EventType]
public record ReactorOrderPlaced(string CustomerEmail, decimal TotalAmount);

import io.cratis.chronicle.events.EventType

@EventType(id = "ReactorOrderPlaced")
data class ReactorOrderPlaced(
    val customerEmail: String,
    val totalAmount: Double
)

import io.cratis.chronicle.events.EventType;

@EventType(id = "ReactorOrderPlaced")
record ReactorOrderPlaced(String customerEmail, double totalAmount) {}

defmodule MyApp.Events.ReactorOrderPlaced do
  use Chronicle.Events.EventType, id: "reactor-order-placed-v1"

  defstruct [:customer_email, :total_amount]
end

import { eventType } from '@cratis/chronicle';

@eventType()
class ReactorOrderPlaced {
    constructor(
        readonly customerEmail: string,
        readonly totalAmount: number
    ) {}
}

Define the reactor

The client SDK decides how a reactor is declared. The common shape is the same: mark a type or module as a reactor, add a handler for the event, and accept event context when the side effect needs metadata such as sequence number, occurrence time, event store, namespace, or correlation.

public interface ReactorEmailGateway
{
    Task SendOrderPlaced(string email, decimal amount, DateTimeOffset occurred);
}

public class OrderNotificationsReactor(ReactorEmailGateway emailGateway) : IReactor
{
    public Task Placed(ReactorOrderPlaced @event, EventContext context) =>
        emailGateway.SendOrderPlaced(
            @event.CustomerEmail,
            @event.TotalAmount,
            context.Occurred);
}

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.observation.Reactor
import java.time.Instant

interface ReactorEmailGateway {
    fun sendOrderPlaced(email: String, amount: Double, occurred: Instant)
}

@Reactor
class OrderNotificationsReactor(private val emailGateway: ReactorEmailGateway) {
    fun placed(event: ReactorOrderPlaced, context: EventContext) {
        emailGateway.sendOrderPlaced(
            event.customerEmail,
            event.totalAmount,
            context.occurred)
    }
}

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.observation.Reactor;
import java.time.Instant;

interface ReactorEmailGateway {
    void sendOrderPlaced(String email, double amount, Instant occurred);
}

@Reactor
class OrderNotificationsReactor {
    private final ReactorEmailGateway emailGateway;

    OrderNotificationsReactor(ReactorEmailGateway emailGateway) {
        this.emailGateway = emailGateway;
    }

    void placed(ReactorOrderPlaced event, EventContext context) {
        emailGateway.sendOrderPlaced(
            event.customerEmail(),
            event.totalAmount(),
            context.getOccurred());
    }
}

defmodule MyApp.EmailGateway do
  def order_placed(_email, _amount, _occurred), do: :ok
end

defmodule MyApp.Reactors.OrderNotificationsReactor do
  use Chronicle.Reactors.Reactor

  alias MyApp.Events.ReactorOrderPlaced

  @handles ReactorOrderPlaced

  @impl true
  def handle(%ReactorOrderPlaced{} = event, context) do
    MyApp.EmailGateway.order_placed(
      event.customer_email,
      event.total_amount,
      Map.get(context, :occurred))

    :ok
  end
end

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

interface ReactorEmailGateway {
    sendOrderPlaced(email: string, amount: number, occurred: Date): Promise<void>;
}

@reactor()
class OrderNotificationsReactor {
    constructor(private readonly emailGateway: ReactorEmailGateway) {}

    // Method name must be the exact camelCase of the event's class name -
    // Chronicle discovers handlers by name, not by parameter type.
    async reactorOrderPlaced(event: ReactorOrderPlaced, context: EventContext): Promise<void> {
        await this.emailGateway.sendOrderPlaced(
            event.customerEmail,
            event.totalAmount,
            context.occurred);
    }
}

Register the reactor

Registration is client-specific. Some clients discover reactor types and register them as part of event-store startup. Others register a reactor instance or include the reactor module in client startup options.

public class ReactorRegistration
{
    public Task Register(IEventStore eventStore) => eventStore.Reactors.Register();
}

import io.cratis.chronicle.IEventStore
import kotlinx.coroutines.Job

class ReactorRegistration(private val emailGateway: ReactorEmailGateway) {
    suspend fun register(store: IEventStore): Job =
        store.reactors.register(OrderNotificationsReactor(emailGateway))
}

import io.cratis.chronicle.IEventStore;
import kotlin.coroutines.Continuation;
import kotlin.coroutines.EmptyCoroutineContext;
import kotlinx.coroutines.BuildersKt;
import kotlinx.coroutines.Job;

class ReactorRegistration {
    private final ReactorEmailGateway emailGateway;

    ReactorRegistration(ReactorEmailGateway emailGateway) {
        this.emailGateway = emailGateway;
    }

    Job register(IEventStore store) throws InterruptedException {
        return (Job) BuildersKt.runBlocking(
            EmptyCoroutineContext.INSTANCE,
            (scope, continuation) -> {
                @SuppressWarnings("unchecked")
                var registerContinuation = (Continuation<? super Job>) continuation;
                return store.getReactors().register(
                    new OrderNotificationsReactor(emailGateway),
                    registerContinuation);
            });
    }
}

defmodule MyApp.ReactorClientConfig do
  def child_spec do
    {Chronicle.Client,
     connection_string: "chronicle://localhost:35000",
     event_store: "store",
     reactors: [MyApp.Reactors.OrderNotificationsReactor]}
  end
end

import { IEventStore } from '@cratis/chronicle';

class ReactorRegistration {
    async register(store: IEventStore): Promise<void> {
        await store.reactors.register();
    }
}

Use a stable reactor ID

By default, clients derive a reactor identifier from the type or module name. Set an explicit ID when the reactor’s identity must survive a rename. Changing the ID makes Chronicle treat it as a different observer, with a different observation position.

[Reactor(id: "order-notifications")]
public class NamedOrderNotificationsReactor : IReactor
{
    public Task Placed(ReactorOrderPlaced @event) => Task.CompletedTask;
}

import io.cratis.chronicle.observation.Reactor

@Reactor(id = "order-notifications")
class NamedOrderNotificationsReactor {
    fun placed(event: ReactorOrderPlaced) {
        // Perform the side effect.
    }
}

import io.cratis.chronicle.observation.Reactor;

@Reactor(id = "order-notifications")
class NamedOrderNotificationsReactor {
    void placed(ReactorOrderPlaced event) {
        // Perform the side effect.
    }
}

defmodule MyApp.Reactors.NamedOrderNotificationsReactor do
  use Chronicle.Reactors.Reactor, id: "order-notifications"

  alias MyApp.Events.ReactorOrderPlaced

  @handles ReactorOrderPlaced

  @impl true
  def handle(%ReactorOrderPlaced{}, _context), do: :ok
end

import { reactor } from '@cratis/chronicle';

@reactor('order-notifications')
class NamedOrderNotificationsReactor {
    // Method name must be the exact camelCase of the event's class name -
    // Chronicle discovers handlers by name, not by parameter type.
    async reactorOrderPlaced(_event: ReactorOrderPlaced): Promise<void> {
        // Perform the side effect.
    }
}

Design for replay

Reactors may see the same event again after replay, recovery, reconnect, or a retry. Make every side effect idempotent. Common patterns are idempotency keys, check-then-act storage, database upserts, or downstream APIs that tolerate duplicate requests.

Next steps

Learn the shared reactor event-processing model
Compare reactors with reducers and projections
Use tagging reactors when you need to organize observer behavior