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.

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.

C#

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

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

Kotlin

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

Java

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

Elixir

defmodule MyApp.ReadModels.WatchingBasicOrder do
  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

TypeScript

for await (const changeset of store.readModels.watch(Order)) {
    if (changeset.removed) {
        continue;
    }

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

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.

C#

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

Kotlin

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

Java

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

Elixir

defmodule MyApp.ReadModels.WatchingFilteringOrder do
  defstruct [:id, :total_amount]
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

TypeScript

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

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

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.

Related topics

• 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