Observing Append Operations

IEventSequence exposes an AppendOperations observable that emits after every Append or AppendMany call on that sequence. Any code can subscribe to this observable to react to appended events in real time — without polling the event log.

AppendOperations

using Cratis.Chronicle.EventSequences;

public interface IObservingAppendsEventSequence
{
    IObservable<IEnumerable<AppendedEventWithResult>> AppendOperations { get; }
}

import io.cratis.chronicle.eventSequences.AppendedEventWithResult
import kotlinx.coroutines.flow.SharedFlow

/**
 * The shape of [io.cratis.chronicle.eventSequences.IEventSequence.appendOperations] - a hot
 * [SharedFlow] that emits after every [io.cratis.chronicle.eventSequences.IEventSequence.append]
 * or [io.cratis.chronicle.eventSequences.IEventSequence.appendMany] call made through that instance.
 */
interface EventSequenceAppendOperationsShape {
    val appendOperations: SharedFlow<List<AppendedEventWithResult>>
}

import io.cratis.chronicle.eventSequences.AppendedEventWithResult;
import io.cratis.chronicle.eventSequences.IEventLog;

import java.util.List;
import java.util.function.Consumer;

import kotlinx.coroutines.Job;

// The shape of the Java bridge for observing append operations: EventLogJavaBridge.watchAppendOperations
// subscribes a callback that receives a list of AppendedEventWithResult after every append or
// appendMany call made through the event log, returning the Job backing the subscription.
interface EventSequenceAppendOperationsShape {
    Job watchAppendOperations(IEventLog eventLog, Consumer<List<AppendedEventWithResult>> callback);
}

Elixir does not support this workflow yet.

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

// eventLog.appendOperations is an AsyncIterable<AppendedEventWithResult[]> - a hot,
// multicast stream: iterating it yields every batch of events appended through this
// specific event log instance from the moment you start iterating (never for
// transactional appends, and never replayed for a late subscriber).
type ObservingAppendsShape = AsyncIterable<AppendedEventWithResult[]>;

The observable emits a collection of AppendedEventWithResult after each operation:

A single-event Append emits a collection containing one element.
A batch AppendMany emits the full batch as one collection.

Each AppendedEventWithResult pairs the appended event with its result:

Member	Type	Description
`Event`	`AppendedEvent`	The appended event, including context and deserialized content
`Event.Content`	`object`	The deserialized event object
`Event.Context`	`EventContext`	Metadata: event source, sequence number, correlation ID, causation chain
`Result`	`AppendResult`	Success flag, sequence number, violations, and errors

Subscribers receive the notification after the operation has completed, whether it succeeded or failed. This observable does not fire for transactional appends through ITransactionalEventSequence.

Subscribing

Inject IEventSequence (or IEventLog) and subscribe to AppendOperations:

using Cratis.Chronicle.EventSequences;

public class ObservingAppendsMonitor(IEventLog eventLog) : IDisposable
{
    readonly IDisposable _subscription = eventLog.AppendOperations.Subscribe(OnAppended);

    static void OnAppended(IEnumerable<AppendedEventWithResult> operations)
    {
        foreach (var item in operations)
        {
            Console.WriteLine($"Event {item.Event.Content.GetType().Name} appended: success={item.Result.IsSuccess}");
        }
    }

    public void Dispose() => _subscription.Dispose();
}

import io.cratis.chronicle.eventSequences.IEventLog
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.launch

/**
 * Collects append operations for as long as [scope] is active, logging each appended event's
 * outcome.
 */
class AppendMonitor(eventLog: IEventLog, scope: CoroutineScope) {
    init {
        scope.launch {
            eventLog.appendOperations.collect { operations ->
                operations.forEach { println("Event ${it.event::class.simpleName} appended: success=${it.result.isSuccess}") }
            }
        }
    }
}

import io.cratis.chronicle.eventSequences.IEventLog;

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

class EventsObservingAppendsSubscribing {
    // Subscribes to append operations for the lifetime of the returned Job; cancel it to stop.
    Job subscribe(IEventLog eventLog) {
        return EventLogJavaBridge.watchAppendOperations(eventLog, operations ->
            operations.forEach(operation ->
                System.out.println("Event " + operation.getEvent().getClass().getSimpleName() +
                    " appended: success=" + operation.getResult().isSuccess())));
    }
}

Elixir does not support this workflow yet.

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

async function monitorAppends(eventLog: IEventLog): Promise<void> {
    for await (const operations of eventLog.appendOperations) {
        for (const item of operations) {
            console.log(`Event ${item.event.eventType.id.value} appended: success=${item.result.isSuccess}`);
        }
    }
}

Always dispose the subscription when you no longer need it to avoid resource leaks.

Integration Testing

The most common use of AppendOperations in application code is through the IEventAppendCollection helper provided by Cratis.Chronicle.XUnit.Integration. It subscribes internally and provides a ready-to-assert collection of AppendedEventWithResult entries. See Chronicle.Testing.EventAppendCollection for full details.

Waiting for Observer Completion After Append

When you need to wait until observers affected by an append operation have completed, use WaitForCompletion() from the Cratis.Chronicle.Observation namespace.

If any affected observer fails while catching up, the returned result includes failed partitions.

Waiting only means something when there are observers behind the connection. The in-process testing surfaces (EventScenario and EventStoreForTesting) run the event sequence without a silo, so an append there never reaches a projection, reducer, or reactor. Rather than report that everything completed, WaitForCompletion() throws CannotWaitForObserverCompletion on an append result that carries no observer surface — assert on the scenario’s own surface, or move the check to an out-of-process integration spec where observers genuinely run.

Append

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Observation;

[EventType]
public record ObservingAppendsSomeEvent(string Data);

public class ObservingAppendsCompletionWaiter(IEventLog eventLog)
{
    public async Task AppendAndWait(EventSourceId eventSourceId)
    {
        var appendResult = await eventLog.Append(eventSourceId, new ObservingAppendsSomeEvent("example"));
        var completion = await appendResult.WaitForCompletion();

        if (!completion.IsSuccess)
        {
            foreach (var failedPartition in completion.FailedPartitions)
            {
                Console.WriteLine($"Observer {failedPartition.ObserverId} failed partition {failedPartition.Partition}");
            }
        }
    }
}

Kotlin does not support this workflow yet.

Java does not support this workflow yet.

defmodule MyApp.Events.ObservingAppendsSomeEvent do
  use Chronicle.Events.EventType, id: "observing-appends-some-event"

  defstruct [:data]
end

defmodule MyApp.ObservingAppendsCompletionWaiter do
  alias Chronicle.EventSequences.EventLog
  alias MyApp.Events.ObservingAppendsSomeEvent

  def append_and_wait(event_source_id) do
    case EventLog.append_and_wait_for_completion(event_source_id, %ObservingAppendsSomeEvent{
           data: "example"
         }) do
      {:ok, %{success: true}} ->
        :ok

      {:ok, %{success: false, failed_partitions: failed_partitions}} ->
        Enum.each(failed_partitions, fn failed_partition ->
          IO.puts("Observer failed partition: #{inspect(failed_partition)}")
        end)

      {:error, reason} ->
        {:error, reason}
    end
  end
end

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

@eventType()
class ObservingAppendsSomeEvent {
    constructor(readonly data: string = '') {}
}

async function appendAndWait(eventLog: IEventLog, eventSourceId: string): Promise<void> {
    const appendResult = await eventLog.append(eventSourceId, new ObservingAppendsSomeEvent('example'));
    const completion = await appendResult.waitForCompletion();

    if (!completion.isSuccess) {
        for (const failedPartition of completion.failedPartitions) {
            console.log(`Observer ${failedPartition.observerId} failed partition ${failedPartition.partition}`);
        }
    }
}

AppendMany

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Observation;

[EventType]
public record ObservingAppendsFirstEvent(string Data);

[EventType]
public record ObservingAppendsSecondEvent(string Data);

public class ObservingAppendsBatchCompletionWaiter(IEventLog eventLog)
{
    public async Task AppendManyAndWait(EventSourceId eventSourceId)
    {
        var appendManyResult = await eventLog.AppendMany(eventSourceId, new object[]
        {
            new ObservingAppendsFirstEvent("first"),
            new ObservingAppendsSecondEvent("second")
        });

        var completion = await appendManyResult.WaitForCompletion();
        if (!completion.IsSuccess)
        {
            // Inspect failed partitions from affected observers
        }
    }
}

Kotlin does not support this workflow yet.

Java does not support this workflow yet.

Elixir does not support this workflow yet.

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

@eventType()
class ObservingAppendsFirstEvent {
    constructor(readonly data: string = '') {}
}

@eventType()
class ObservingAppendsSecondEvent {
    constructor(readonly data: string = '') {}
}

async function appendManyAndWait(eventLog: IEventLog, eventSourceId: string): Promise<void> {
    const appendManyResults = await eventLog.appendMany(eventSourceId, [
        new ObservingAppendsFirstEvent('first'),
        new ObservingAppendsSecondEvent('second')
    ]);

    for (const appendResult of appendManyResults) {
        const completion = await appendResult.waitForCompletion();
        if (!completion.isSuccess) {
            // Inspect completion.failedPartitions from affected observers.
        }
    }
}