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.

Kotlin Flow

Declare the observable method on the companion object and annotate it with @JvmStatic so that Java callers and generated performers can reach it:

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

repository.observeAll() may return any cold or hot Flow. MutableStateFlow is the JVM substitute for .NET’s ISubject<T> when you need a push-side handle and an observable StateFlow for HTTP snapshot support — it requires no external dependency:

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

Java Flow.Publisher

Java static methods may return java.util.concurrent.Flow.Publisher<T>. Arc wraps it with asKotlinFlow() before the observable pipeline:

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

java.util.concurrent.SubmissionPublisher<T> gives you a push-side handle with no external dependency:

private static final java.util.concurrent.SubmissionPublisher<List<TaskView>> publisher =
    new java.util.concurrent.SubmissionPublisher<>();

public static java.util.concurrent.Flow.Publisher<List<TaskView>> all() {
    return publisher;
}

Observable state a snapshot can read

A SubmissionPublisher emits, but it holds nothing. Nobody who subscribes learns the current value until the next change — and that is visible on the wire, because Arc answers a snapshot GET from the source’s current value and reports a source without one as 202 Not Ready rather than holding the request open.

Kotlin has MutableStateFlow for this. The JDK has nothing equivalent, so Arc supplies ObservableState<T>:

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

It 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. Kotlin code has no reason to: use MutableStateFlow and return a Flow.

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.