Watching Read Models

Watching a read model gives a client a stream of changes as Chronicle applies projections and reducers. Use it for live dashboards, cache invalidation, background coordination, and interfaces that should update without polling.

Not every Chronicle client exposes the watch API yet. The examples below show the clients that currently do.

For a list that needs an initial snapshot and a complete replacement set after changes, use Observing Read Models instead. Watch emits individual changesets, not the initial collection.

What a change contains

Each change tells you:

Field	Meaning
Namespace	The Chronicle namespace where the change happened
Key	The read model key that changed
Read model	The current state after the change
Removed	Whether the instance was removed

The change carries the current read model state, not the previous state. If you need to detect a transition, keep the last value you saw for each key or react to the domain event with a reactor.

Watch a read model

The read model type must be registered before you subscribe.

using var subscription = eventStore.ReadModels
    .Watch<Order>()
    .Subscribe(changeset =>
    {
        if (changeset.Removed || changeset.ReadModel is null)
        {
            return;
        }

        Console.WriteLine($"{changeset.ModelKey}: {changeset.ReadModel.Status}");
    });

View C# snippet source on GitHub

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.readModels.ReadModel
import kotlinx.coroutines.flow.collect

@ReadModel
data class WatchingBasicOrder(val status: String = "", val totalAmount: Double = 0.0)

suspend fun watchOrders(store: IEventStore) {
    store.readModels.watch(WatchingBasicOrder::class).collect { changeset ->
        if (changeset.removed || changeset.readModel == null) return@collect

        println("${changeset.modelKey}: ${changeset.readModel!!.status}")
    }
}

View Kotlin snippet source on GitHub

import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.readModels.ReadModel;

import io.cratis.chronicle.java.ReadModelsJavaBridge;
import kotlinx.coroutines.Job;

@ReadModel
class WatchingBasicOrder {
    private String status = "";
    private double totalAmount = 0;

    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }

    public double getTotalAmount() { return totalAmount; }
    public void setTotalAmount(double totalAmount) { this.totalAmount = totalAmount; }
}

class ReadModelsWatchingReadModelsBasic {
    Job watchOrders(EventStore store) {
        return ReadModelsJavaBridge.watch(store.getReadModels(), WatchingBasicOrder.class, changeset -> {
            if (changeset.getRemoved() || changeset.getReadModel() == null) {
                return;
            }

            System.out.println(changeset.getModelKey() + ": " + changeset.getReadModel().getStatus());
        });
    }
}

View Java snippet source on GitHub

defmodule MyApp.ReadModels.WatchingBasicOrder do
  use Chronicle.ReadModels.ReadModel

  defstruct id: "", status: ""
end

defmodule MyApp.WatchingReadModelsBasicMonitor do
  alias MyApp.ReadModels.WatchingBasicOrder

  def start_watching do
    {:ok, watcher} = Chronicle.ReadModels.watch(WatchingBasicOrder)
    watcher
  end

  def handle_next_change do
    receive do
      {:chronicle_read_model_changed, WatchingBasicOrder, changeset} ->
        if changeset.removed or is_nil(changeset.read_model) do
          :ok
        else
          IO.puts("#{changeset.model_key}: #{changeset.read_model.status}")
        end
    end
  end

  def stop_watching(watcher), do: Chronicle.ReadModels.unwatch(watcher)
end

View Elixir snippet source on GitHub

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

const watcher = store.readModels.createWatcher(Order, { signal: AbortSignal.timeout(60_000) });
const stopRefreshing = watcher.onResubscribed(async () => {
    // Requery the entire collection, including additions and removals missed during the outage.
    const orders = await store.readModels.getInstances(Order);
    console.table(orders);
});
try {
    await watcher.subscribed;
    // The kernel is now subscribed; events appended after this point can be observed.
    for await (const changeset of watcher) {
        console.log(changeset.changeType === ReadModelChangeType.Added ? 'Added' : 'Changed',
            changeset.key, changeset.changeContext?.correlationId);
        if (!changeset.removed) {
            console.log(changeset.readModel.status);
        }
    }
} finally {
    stopRefreshing();
    watcher.dispose();
}

createWatcher() starts immediately. Await subscribed before producing events when you need to observe their changes. The example stops after one minute; a service can pass its shutdown signal instead. watch() keeps its original async-iterable return type and lazy, pull-based behavior: the stream starts on the first next(), and each change waits for consumer demand. It has no client-side buffer or overflow limit. Use createWatcher() for readiness and lifetime controls.

By default, transport errors reject iteration with the original error and stream completion ends iteration. Registering the onResubscribed callback opts this example into resumption: transport failures (gRPC 1/4/13/14) and stream completion resume on the same watcher. You can also opt in explicitly with createWatcher(Order, { resume: true }), but must still account for missed changes. onResubscribed fires after the next subscription acknowledgment, not the first, so you can refresh even if no further changes arrive. Register before awaiting readiness; late registrations do not replay earlier notifications. Async callbacks are awaited before reading further changes on the resumed stream; a thrown error or rejected promise fails iteration. The returned function unregisters the callback, as does disposal. Removing the last callback disables resumption for subsequent failures unless resume: true was passed.

After a disconnect, watcher.subscribed waits for the next acknowledgment; previously resolved promises remain resolved. Changes received before the outage are still delivered, but changes during the outage are not reported. The callback refreshes the collection while the loop logs received changes; if you maintain a displayed collection, treat those changes as invalidations and requery rather than overwriting refreshed state with older buffered payloads. Non-transport stream errors, conversion/compliance failures, and callback failures reject iteration and require a new watcher. A terminal connection failure requires a new client after correcting its cause.

For createWatcher(), stream reads pause after acknowledgment at 1,024 buffered changes until the consumer pulls. Before each acknowledgment, the watcher must keep reading to reach the marker; exceeding 1,024 buffered changes then fails iteration and readiness. Use createWatcher(Order, { maxBuffered: 256 }) only if you want overflow failure instead of backpressure; the positive-integer limit applies before and after acknowledgment. Both APIs yield optional changeType and changeContext fields.

View TypeScript snippet source on GitHub

Dispose, cancel, or break out of the subscription when the caller no longer needs updates. Long-lived subscriptions should be owned by a service with an explicit lifetime.

Filter changes

Filter as close to the subscription as the client allows. This keeps application code focused on the state changes it actually cares about.

using var subscription = eventStore.ReadModels
    .Watch<Order>()
    .Where(changeset => changeset.ReadModel?.TotalAmount > threshold)
    .Subscribe(changeset =>
    {
        Console.WriteLine($"{changeset.ModelKey}: {changeset.ReadModel!.TotalAmount:C}");
    });

View C# snippet source on GitHub

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.readModels.ReadModel
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.filter

@ReadModel
data class WatchingFilteringOrder(val status: String = "", val totalAmount: Double = 0.0)

/**
 * Filtering happens client-side - the server still sends every change for the read model type.
 */
suspend fun watchHighValueOrders(store: IEventStore, threshold: Double) {
    store.readModels.watch(WatchingFilteringOrder::class)
        .filter { (it.readModel?.totalAmount ?: 0.0) > threshold }
        .collect { changeset ->
            println("${changeset.modelKey}: ${changeset.readModel?.totalAmount}")
        }
}

View Kotlin snippet source on GitHub

import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.readModels.ReadModel;

import io.cratis.chronicle.java.ReadModelsJavaBridge;
import kotlinx.coroutines.Job;

@ReadModel
class WatchingFilteringOrder {
    private String status = "";
    private double totalAmount = 0;

    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }

    public double getTotalAmount() { return totalAmount; }
    public void setTotalAmount(double totalAmount) { this.totalAmount = totalAmount; }
}

class ReadModelsWatchingReadModelsFiltering {
    // Filtering happens client-side, inside the callback - the server still sends every change
    // for the read model type.
    Job watchHighValueOrders(EventStore store, double threshold) {
        return ReadModelsJavaBridge.watch(store.getReadModels(), WatchingFilteringOrder.class, changeset -> {
            WatchingFilteringOrder order = changeset.getReadModel();
            if (order == null || order.getTotalAmount() <= threshold) {
                return;
            }

            System.out.println(changeset.getModelKey() + ": " + order.getTotalAmount());
        });
    }
}

View Java snippet source on GitHub

defmodule MyApp.ReadModels.WatchingFilteringOrder do
  use Chronicle.ReadModels.ReadModel

  defstruct id: "", total_amount: 0.0
end

defmodule MyApp.WatchingReadModelsFilteringMonitor do
  alias MyApp.ReadModels.WatchingFilteringOrder

  def start_watching do
    {:ok, watcher} = Chronicle.ReadModels.watch(WatchingFilteringOrder)
    watcher
  end

  def handle_next_change(threshold) do
    # Filtering happens client-side, in the process receiving the changes.
    receive do
      {:chronicle_read_model_changed, WatchingFilteringOrder, changeset} ->
        if changeset.read_model && changeset.read_model.total_amount > threshold do
          IO.puts("#{changeset.model_key}: #{changeset.read_model.total_amount}")
        end
    end
  end
end

View Elixir snippet source on GitHub

for await (const changeset of store.readModels.watch(Order)) {
    if (changeset.readModel.totalAmount <= threshold) {
        continue;
    }

    console.log(`${changeset.key}: ${changeset.readModel.totalAmount}`);
}

View TypeScript snippet source on GitHub

Filtering happens client-side. If the client subscribes to a busy read model, the server still sends matching change notifications for that read model type.

When to use this

Use read-model watching when:

A UI needs pushed updates.
A cache should invalidate or refresh when state changes.
A background service reacts to state changes instead of raw events.
The consumer can manage a long-running subscription.

Prefer current-state reads when:

The caller only needs a value once.
The process cannot own a long-running subscription.
The reaction should be based on the original event rather than derived state.

Getting a Single Instance - Read current state on demand
Getting a Collection of Instances - Replay all current instances
Getting Snapshots - Inspect historical state
Reactors - React to events directly

Watching Read Models

What a change contains

Watch a read model

Filter changes

When to use this

Related topics