Skip to content
Cratis

Observable storage snapshots

Return observable storage snapshots

Section titled “Return observable storage snapshots”

Use storage observations when a query should return current results again as the stored data changes. Each snapshot replaces the previous result; it is not a mutation event for the caller to apply. Both Spring Data integrations provide injectable query services whose methods return Kotlin Flow.

A Flow is cold by default: collection reads the initial snapshot, owns the subscription, and closes it on cancellation. Use observeShared when subscribers should share an observation and replay one complete snapshot while subscribers exist. Java callers can use the demand-aware observePublisher methods or callback overloads, whose returned AutoCloseable cancels collection.

These storage results are separate from the transport’s full or delta transfer mode. Returning a storage snapshot does not require sending every replacement as a full snapshot over the wire.

Observe MongoDB

Section titled “Observe MongoDB”

MongoDB change streams require a replica set or sharded cluster. With that prerequisite in place, inject MongoObservableQuery into a model-bound query. Its observe, observeSingle, and observeById methods re-run the Spring Data query after insert, update, replace, delete, or invalidate notifications: a relevant change leads to another read of the current result.

The Kotlin fences on this page are excerpts, not complete classes. Each method belongs inside the companion object of a @ReadModel class such as TaskView, as in the queries guide.

@JvmStatic
fun observeTasks(
    @FromServices queries: MongoObservableQuery
): Flow<List<TaskView>> = queries.observe(TaskView::class.java)


@JvmStatic
fun observeTask(
    id: String,
    @FromServices queries: MongoObservableQuery
): Flow<TaskView> = queries.observeById(TaskView::class.java, id)

A Query supplies filtering; observeById also narrows change notifications by document key. Updates and replacements trigger a fresh snapshot, so a document that stops matching a filter disappears from the result rather than remaining in a client-maintained list.

Pass the tenant explicitly to the observation method. A custom MongoOperationsResolver can select operations from the captured tenantId; it must not consult thread-local request state. The default resolver uses the application’s single MongoOperations bean, not an automatically selected tenant store.

MongoChangeStreamWatcher is the change-stream SPI. The default ReconnectingMongoChangeStreamWatcher uses SpringDataMongoChangeStreamSource, resumes from the last token after transient failures, applies capped exponential backoff, and closes its cursor when the collector is canceled. Cursor operations run on Dispatchers.IO without a hidden flowOn channel.

MongoObservationOptions.bufferCapacity controls the queued changes exactly: zero keeps the producer and collector in rendezvous, while a positive value permits only that many queued changes.

Kotlin type-inference extensions for MongoDB

Section titled “Kotlin type-inference extensions for MongoDB”

The arc-spring-data-mongodb artifact ships Kotlin extension functions on MongoObservableQuery that infer the document type from the generic parameter, eliminating the explicit ::class.java argument. They are top-level functions, so import them by name, for example import io.cratis.arc.springdata.mongodb.observe and import io.cratis.arc.springdata.mongodb.observeById:

@JvmStatic
fun observeTasks(
    @FromServices queries: MongoObservableQuery
): Flow<List<TaskView>> = queries.observe<TaskView>()


@JvmStatic
fun observeTask(
    id: String,
    @FromServices queries: MongoObservableQuery
): Flow<TaskView> = queries.observeById<TaskView>(id)

For criteria-based filtering, the extension wraps the criteria in a Query for you. The two language tabs below request the same filtered snapshot:

public static Flow.Publisher<List<TaskView>> observeActiveTasks(@FromServices MongoObservableQuery queries) {
    return MongoObservations.observe(queries, TaskView.class, Criteria.where("active").is(true));
}

View Java snippet source on GitHub

The same extensions cover observeList, observeSingle, observeById, and observeShared; each delegates to the corresponding MongoObservableQuery method.

Call these extensions on Arc’s own MongoObservableQuery. They cannot be placed on Spring Data’s MongoCollection<T>, repository interfaces, or any other foreign type; all observe* extensions belong to the Arc query service.

Java convenience access for MongoDB

Section titled “Java convenience access for MongoDB”

Kotlin extensions are invisible from Java. The MongoObservations object provides @JvmStatic Criteria-based helpers that Java callers can use without manually constructing a Query, as the Java tab above shows.

Each MongoObservations helper wraps the Criteria into a Query and delegates to the matching MongoObservableQuery method. For Query-based and callback access, call the @JvmOverloads methods on MongoObservableQuery directly.

The tenant overload is also available:

MongoObservations.observe(queries, TaskView.class, Criteria.where("active").is(true), tenantId);

Observe JPA

Section titled “Observe JPA”

Inject JpaObservableQuery and return observe, observeList, observeSingle, or observeById from the read-model query. The default list query uses the mapped JPA entity name. Supply a JpaSnapshotQuery when the snapshot needs predicates, ordering, or fetch joins; the callback is also usable from Java.

The tab below defines the read side. Unlike MongoDB change streams, JPA needs the application to connect successful writes to explicit notifications.

public static Flow.Publisher<List<TaskView>> observeTasks(@FromServices JpaObservableQuery queries) {
    return JpaObservations.observePublisher(queries, TaskView.class);
}

View Java snippet source on GitHub

On the write side, perform the persistence operation and explicitly call the existing DatabaseChangePublisher.publish(TaskView::class.java, tenantId) contract for the observed entity class and matching tenant ID. Returning the observation or injecting a repository does not install a database-wide write hook. Arrange publication so a notification can cause a read of a successful, visible write; transaction enrollment and notification wiring are separate concerns.

With active Spring transaction synchronization, TransactionAwareDatabaseChangeNotifier coalesces repeated entity-class/tenant-ID pairs per notifier instance and attempts to emit them from afterCommit. Its afterCompletion callback clears the pending pairs, including on rollback completion. Without active synchronization, publication immediately attempts tryEmit rather than waiting for commit: call it only after the write has succeeded and is visible. Publishing does not create a transaction or guarantee notification delivery.

The notification must match both the observed entity class and the tenant ID. That tenant label filters notifications; it does not select storage. A subsequent snapshot still reads through the query service’s configured entity-manager factory, not a persistence unit selected by the label.

The built-in notifier is process-local, with no replay. Applications needing cross-process notifications must replace the DatabaseChangeNotifier bean with a database-native implementation and expose a corresponding DatabaseChangePublisher where local writes also need publishing. Arc does not silently poll either store; polling must be an application-owned, explicitly configured notifier.

JpaObservationOptions.bufferCapacity must be positive and bounds pending invalidations. When full, it replaces the oldest invalidation: these are requests to read current state again, not a durable record of each write. Collection reads the initial snapshot before starting notification collection. Later notifications are debounce-coalesced, and equal snapshots are suppressed, so do not rely on observing every write or on gap-free subscription startup. Snapshot delivery itself uses a rendezvous handoff rather than flowOn’s implicit buffer.

Kotlin type-inference extensions for JPA

Section titled “Kotlin type-inference extensions for JPA”

The arc-spring-data-jpa artifact ships Kotlin extension functions on JpaObservableQuery that infer the entity type from the generic parameter. Import them by name from io.cratis.arc.springdata.jpa, for example import io.cratis.arc.springdata.jpa.observe:

@JvmStatic
fun observeTasks(
    @FromServices queries: JpaObservableQuery
): Flow<List<TaskView>> = queries.observe<TaskView>()


@JvmStatic
fun observeTask(
    id: String,
    @FromServices queries: JpaObservableQuery
): Flow<TaskView> = queries.observeById<TaskView>(id)

A custom snapshot query can be supplied as a trailing lambda:

fun observeActiveTasks(
    @FromServices queries: JpaObservableQuery
): Flow<List<TaskView>> =
    queries.observe<TaskView>(
        query = JpaSnapshotQuery { em ->
            em.createQuery(
                "select t from TaskView t where t.active = true",
                TaskView::class.java
            ).resultList
        }
    )

The same extensions cover observeList, observeSingle, observeById, and observeShared.

As with the MongoDB helpers, all observe* extensions are on Arc’s own JpaObservableQuery. They cannot be placed on Spring Data’s JpaRepository or any other foreign interface.

Java convenience access for JPA

Section titled “Java convenience access for JPA”

Kotlin extensions are invisible from Java. The JpaObservations object provides @JvmStatic helpers for the two entry points that have no default query in the base API:

  • JpaObservations.observePublisher(queries, entityType) — returns a demand-aware Flow.Publisher<List<T>> using the default select-all JPQL query.
  • JpaObservations.observeShared(scope, queries, entityType) — returns a SharedFlow<List<T>> that replays the latest snapshot while subscribers exist.

Both helpers supply the same default JPQL query (select entity from <EntityName> entity) as the Kotlin extensions. For a custom snapshot query or callback access, call the @JvmOverloads methods on JpaObservableQuery directly.

Current public-seam boundary

Section titled “Current public-seam boundary”

The integrations use public Arc and Spring contracts for contextual command read-model parameters, deterministic ownership, generated QueryRequest/QueryContext injection, exact Spring Data Commons Pageable/Sort parameters, and exact Page<T> response normalization. The store-specific request and page adapters remain compatibility utilities, not request-scoped beans.

Choose the access path that actually selects the store. Repository injection is ordinary Spring dependency injection and is not automatically tenant-routed by Arc. JPA observable snapshots use their configured entity-manager factory; tenant labels on notifications are not storage isolation. Mongo observable paths use MongoOperationsResolver, including the certified adapter when a tenant-aware resolver is supplied.

Transaction enrollment is a separate decision again. Imperative transaction scopes remain explicit fixed-store opt-ins and are not coroutine-safe; observing a query does not remove those restrictions.

Community: Discord