Declare observable queries

Choose a return type

An observable query is a static method on a @ReadModel companion object (Kotlin) or a Java static method that returns a streaming type. Arc recognises three families:

• Kotlin kotlinx.coroutines.flow.Flow<T> or Flow<List<T>>
• JDK java.util.concurrent.Flow.Publisher<T> or Publisher<List<T>>
• RxJava 3 Observable<T>, ObservableSource<T>, or Subject<T> — requires the optional arc-rxjava3 artifact at runtime (see RxJava 3 option)

Arc generates an OBSERVABLE performer for any of these. One-shot returns (T, List<T>, or Page<T>) stay request-response.

Return a stream

Declare the observable query as a static method — on the companion object with @JvmStatic in Kotlin, as a static method in Java — so generated performers can reach it. Kotlin returns a Flow; Java returns a java.util.concurrent.Flow.Publisher, which Arc wraps with asKotlinFlow() before the observable pipeline:

Java

@ReadModel
@AllowAnonymous
public record TaskView(String id, String title) {
    public static java.util.concurrent.Flow.Publisher<List<TaskView>> all(
        @FromServices TaskRepository repository
    ) {
        return repository.observeAll();
    }
}

View Java snippet source on GitHub

Kotlin

@ReadModel
@AllowAnonymous
data class TaskView(val id: String, val title: String) {
    companion object {
        @JvmStatic
        fun all(@FromServices repository: TaskRepository): Flow<List<TaskView>> =
            repository.observeAll()
    }
}

View Kotlin snippet source on GitHub

repository.observeAll() may return any cold or hot stream.

Keep state a snapshot can read

A source that only emits holds nothing. A JDK SubmissionPublisher, for example, tells a new subscriber nothing until the next change — and that is visible on the wire, because Arc answers a snapshot GET from the source’s current value and, by default, reports a source without one as 202 Not Ready rather than holding the request open. A caller can pass waitForFirstResult=true to wait up to the configured limit instead.

Hold the state instead. Kotlin uses MutableStateFlow, the JVM substitute for .NET’s ISubject<T>. The JDK has no equivalent, so Arc supplies ObservableState<T> for Java. Neither needs an external dependency:

Java

@ReadModel
@AllowAnonymous
public record TaskView(String id, String title) {
    private static final ObservableState<List<TaskView>> tasks = new ObservableState<>(List.of());

    public static Flow.Publisher<List<TaskView>> all() {
        return tasks;
    }
}

View Java snippet source on GitHub

Kotlin

private val _tasks = MutableStateFlow<List<TaskView>>(emptyList())

@ReadModel
@AllowAnonymous
data class TaskView(val id: String, val title: String) {
    companion object {
        @JvmStatic
        fun all(): Flow<List<TaskView>> = _tasks
    }
}

View Kotlin snippet source on GitHub

The same holds when the state lives in a service rather than on the read model:

Java

@Component
public final class TaskSource {
    private final ObservableState<List<TaskView>> tasks = new ObservableState<>(List.of());

    public Flow.Publisher<List<TaskView>> observe() {
        return tasks;
    }

    public void publish(List<TaskView> updated) {
        tasks.set(updated);
    }
}

View Java snippet source on GitHub

Kotlin

@Component
class TaskSource {
    private val tasks = MutableStateFlow<List<TaskView>>(emptyList())

    fun observe(): Flow<List<TaskView>> = tasks

    fun publish(updated: List<TaskView>) {
        tasks.value = updated
    }
}

View Kotlin snippet source on GitHub

ObservableState<T> is a Flow.Publisher<T>, so a query method returns it directly. Every subscriber sees the current value first and then each change; a subscriber that falls behind sees only the newest value rather than a backlog, which is the right behavior for state. Reach for it whenever a Java observable query should also answer a plain GET.

RxJava 3 option

If your application already uses RxJava 3, you may return io.reactivex.rxjava3.core.Observable<T>, ObservableSource<T>, or subjects.Subject<T> from the same static-method shape. Add the optional runtime artifact:

implementation("io.cratis:arc-rxjava3:0.0.0-SNAPSHOT")

Then declare the query exactly as you would for Flow:

@ReadModel
data class TaskView(val id: String, val title: String) {
    companion object {
        @JvmStatic
        fun all(@FromServices repository: TaskRepository): Observable<List<TaskView>> =
            repository.rxObserveAll()
    }
}

Java static methods work the same way:

@ReadModel
public record TaskView(String id, String title) {
    public static io.reactivex.rxjava3.core.Observable<List<TaskView>> all(
        @FromServices TaskRepository repository
    ) {
        return repository.rxObserveAll();
    }
}

Backpressure and buffering. Observable has no backpressure protocol. Arc buffers up to 64 values between the RxJava producer and the Kotlin collector; a producer that outruns the collector beyond that fails the flow with IllegalStateException rather than dropping values silently. Use a backpressure-aware source such as Flowable, or an explicit RxJava operator such as toFlowable, when the producer can outpace the consumer.

Subject as a push handle. io.reactivex.rxjava3.subjects.PublishSubject or BehaviorSubject serve the same push-side role as MutableStateFlow, but require the arc-rxjava3 artifact:

private val subject = io.reactivex.rxjava3.subjects.PublishSubject.create<List<TaskView>>()

companion object {
    @JvmStatic
    fun all(): io.reactivex.rxjava3.subjects.Subject<List<TaskView>> = subject
}

Transfer mode: full snapshots versus deltas

Add transferMode to the subscription body:

• full — every emission carries the complete current snapshot with no change set.
• delta — the first emission is a full snapshot; subsequent emissions carry only the change set.
• Omitting transferMode keeps the legacy behaviour: every emission carries both the full snapshot and a changeSet, with the first one listing every item as added.

For delta mode, Arc needs a stable item identity. When no identity accessor exists, or an extracted key is null or duplicated, Arc falls back to exact serialised JSON identity. That fallback reports additions and removals only — changing any field appears as one removal plus one addition, not a replacement. Use stable, unique item identities when replacements or lower serialisation cost matter.

Change sets do not encode position changes. Do not rely on delta updates to reproduce list reordering.

Subscription lifetime

Each SSE or WebSocket subscription is independently authorised and isolated. The subscription remains open until the source Flow completes, the connection drops, or an Unauthorized guard terminates it. Cancelling collection from the Arc side disposes any underlying RxJava subscription or JDK Flow subscription.

Deliberately rejected types

reactor.core.publisher.Flux is rejected at compile time. Reactor is part of the Spring ecosystem, and the Source module and its artifacts/metadata packages must remain Spring-free — enforced by ./gradlew checkSpringBoundary. Accepting Flux as a query return type would draw a Reactor compile-time dependency into that boundary.

org.reactivestreams.Publisher is also rejected. The org.reactivestreams interfaces are superseded by the structurally identical JDK java.util.concurrent.Flow types introduced in Java 9. Use java.util.concurrent.Flow.Publisher<T> instead.

Use @QueryTransport(QueryTransportType.OBSERVABLE) only when you need to annotate a method that Arc would not otherwise recognise as observable. Annotating a non-streaming return with OBSERVABLE is a compile-time error.