This is an automated email from the ASF dual-hosted git repository.
jamesbognar pushed a commit to branch docs
in repository https://gitbox.apache.org/repos/asf/juneau.git
The following commit(s) were added to refs/heads/docs by this push:
new 44b087fd3b TODO-119/120 - Reactive-streams REST returns (JDK
Flow.Publisher core + Reactor/RxJava bridge) as opt-in
juneau-rest-server-reactive / juneau-rest-server-reactor modules
44b087fd3b is described below
commit 44b087fd3b24a42ca1e0eb6837f720d11c14e1d5
Author: James Bognar <[email protected]>
AuthorDate: Fri May 29 12:31:56 2026 -0400
TODO-119/120 - Reactive-streams REST returns (JDK Flow.Publisher core +
Reactor/RxJava bridge) as opt-in juneau-rest-server-reactive /
juneau-rest-server-reactor modules
---
pages/release-notes/9.5.0.md | 97 ++++++++++++++
pages/topics/10.20m.RestServerReactive.md | 212 ++++++++++++++++++++++++++++++
sidebars.ts | 5 +
3 files changed, 314 insertions(+)
diff --git a/pages/release-notes/9.5.0.md b/pages/release-notes/9.5.0.md
index 5786563f58..84d55f1acc 100644
--- a/pages/release-notes/9.5.0.md
+++ b/pages/release-notes/9.5.0.md
@@ -2685,6 +2685,103 @@ New API:
- **`RestOpContext.getAsyncCompletionExecutor()`** — op-level resolved
executor with resource-level fallback.
- **`RestServerConstants.PROPERTY_asyncCompletionExecutor`** — annotation
property key constant.
+#### Reactive-Streams return types — `Flow.Publisher` core + Reactor / RxJava
bridge (TODO-119 + TODO-120)
+
+`@RestOp` handlers may now return reactive-streams values. **The feature is
fully opt-in by adding a module to the classpath** — a bare
`juneau-rest-server` has zero reactive behavior and `DefaultConfig` wires no
reactive processor. The JDK-native `java.util.concurrent.Flow.Publisher<T>` is
supported by the new dependency-free, opt-in `juneau-rest-server-reactive`
module; Project Reactor (`Mono` / `Flux`), RxJava 3 (`Single` / `Maybe` /
`Completable` / `Flowable` / `Observable`), and the [...]
+
+```java
+@Rest(path="/feed")
+public class FeedResource extends BasicRestServlet {
+
+ // Single-value reactive type — collapses onto the existing async
(CompletableFuture) path.
+ @RestGet("/{id}")
+ public Mono<Article> get(@Path String id) {
+ return articleService.findById(id); // Reactor Mono
→ CompletionStage
+ }
+
+ // Bounded stream, default shape — collected into a JSON array.
+ @RestGet("/recent")
+ public Flux<Article> recent() {
+ return articleService.recent(20); // buffered to
a List, serialized as JSON
+ }
+
+ // Unbounded stream as Server-Sent Events — opt in via the response
content type.
+ @RestGet("/live")
+ public Flux<SseEvent> live(RestResponse res) {
+ res.setContentType("text/event-stream"); // → SSE
frames, one per element
+ return articleService.liveEvents();
+ }
+
+ // Same stream as newline-delimited JSON.
+ @RestGet("/ndjson")
+ public Flux<Article> ndjson(RestResponse res) {
+ res.setContentType("application/x-ndjson"); // → one JSON
object per line
+ return articleService.recent(1000);
+ }
+}
+```
+
+Unified architecture (the two promoted plan items ship as one coherent
feature):
+
+- **Opt-in by classpath presence.** `ReactiveResponseProcessor` is **not**
wired into `DefaultConfig`. It ships in `juneau-rest-server-reactive` with a
`META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor` provider
file; `RestContext` discovers module-contributed response processors via
`ServiceLoader` and front-loads them ahead of `AsyncResponseProcessor`. On a
bare `juneau-rest-server` the discovery finds nothing and the processor chain
is byte-for-byte the pre-feature [...]
+- **`ReactiveResponseProcessor`** (`org.apache.juneau.rest.reactive`, in the
opt-in `juneau-rest-server-reactive` module) is the single shared spine. Once
on the classpath it is front-loaded ahead of `AsyncResponseProcessor` and
natively understands `Flow.Publisher`. Synchronous returns (`String`, beans,
`byte[]`, …) are untouched.
+- **Single-value reactive types** (`Mono`, `Single`, `Maybe`, `Completable`)
are adapted to a `CompletionStage` and collapse onto the existing
`AsyncResponseProcessor` path — inheriting its timeout,
`@Rest(asyncCompletionExecutor)` (TODO-118), and SLF4J MDC bridge (TODO-117)
behavior for free.
+- **Multi-value streams** are rendered in one of three shapes, selected by the
negotiated response media type (handler `Content-Type`, then request `Accept`):
+ - **Buffer** (default) — all elements collected into a `List` and serialized
through the normal chain (e.g. a JSON array). Wrapped in a `CompletableFuture`
so a slow producer never blocks the request thread.
+ - **SSE** (`text/event-stream`) — each element emitted as a
Server-Sent-Events frame; `SseEvent` elements are written verbatim, any other
element type is JSON-encoded into the `data:` field.
+ - **NDJSON** (`application/x-ndjson`, `application/jsonl`) — each element
JSON-encoded on its own line.
+- **Backpressure** — streaming subscribers request one element at a time
(`request(1)` on subscribe and again after each frame is flushed). Because the
servlet output stream blocks until the socket drains, the producer is paced by
the client and the server-side buffer stays bounded.
+- **Third-party libraries are `provided`-scoped** — `reactive-streams`,
`reactor-core`, and `rxjava` are all declared `provided` in
`juneau-rest-server-reactor`. A `dependency:tree` on `juneau-rest-server` never
surfaces them. Adapters are discovered lazily via `ServiceLoader` and skipped
at runtime if their backing library is absent, so a consumer adds
`juneau-rest-server-reactor` plus only the reactive library (and version) they
actually use.
+- **Extensibility** — the `ReactiveStreamsAdapter` SPI + `Adaptation` value
type let third parties bridge any other reactive library to the core
`Flow.Publisher` / `CompletionStage` plumbing without modifying
`juneau-rest-server`.
+
+New API (all in the opt-in `juneau-rest-server-reactive` module):
+
+- **`org.apache.juneau.rest.reactive.ReactiveResponseProcessor`** — the
response processor, auto-registered via `ServiceLoader` when the
`juneau-rest-server-reactive` module is on the classpath (no `DefaultConfig`
wiring).
+- **`org.apache.juneau.rest.reactive.ReactiveStreamsAdapter`** — SPI for
bridging third-party reactive types (`boolean canAdapt(Object)` + `Adaptation
adapt(Object)`).
+- **`org.apache.juneau.rest.reactive.Adaptation`** — single-value
(`CompletionStage`) or streaming (`Flow.Publisher`) adaptation result.
+
+`RestContext` now discovers module-contributed `ResponseProcessor`s via
`ServiceLoader` (inert on a bare classpath) — this is the seam that
auto-activates the reactive processor.
+
+See [Reactive-Streams Returns](/docs/topics/RestServerReactive) for the full
topic.
+
+#### New module: `juneau-rest-server-reactive` (TODO-119/120 refactor)
+
+A new dependency-free, opt-in module that holds the JDK-native
`Flow.Publisher` spine (`ReactiveResponseProcessor`), the
`ReactiveStreamsAdapter` SPI, the `Adaptation` value type, and the
`META-INF/services/...ResponseProcessor` auto-registration file. Adding it
alone enables `Flow.Publisher<T>` returns with no third-party dependency. A
bare `juneau-rest-server` without it has zero reactive behavior.
+
+```xml
+<dependency>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest-server-reactive</artifactId>
+ <version>9.5.0</version>
+</dependency>
+```
+
+#### New module: `juneau-rest-server-reactor` (TODO-119)
+
+A further opt-in bridge module (depends on `juneau-rest-server-reactive`)
providing `ReactiveStreamsAdapter` implementations for the popular reactive
libraries. Server-side only — there is no client-side counterpart. All three
backing libraries are `provided`-scope.
+
+| Adapter | Handles | Adapts to |
+|---|---|---|
+| `ReactorReactiveAdapter` | `Mono<T>` / `Flux<T>` | `Mono` →
`CompletionStage` (via `toFuture()`); `Flux` → `Flow.Publisher` (via
`FlowAdapters`) |
+| `RxJavaReactiveAdapter` | `Single` / `Maybe` / `Completable` / `Flowable` /
`Observable` | single types → `CompletionStage`; `Flowable` → `Flow.Publisher`;
`Observable` → `Flowable(BackpressureStrategy.BUFFER)` → `Flow.Publisher` |
+| `ReactiveStreamsPublisherAdapter` | generic
`org.reactivestreams.Publisher<T>` | `Flow.Publisher` (via `FlowAdapters`);
registered last so `Mono`/`Flux` are matched as single/stream first |
+
+Maven coordinate:
+
+```xml
+<dependency>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest-server-reactor</artifactId>
+ <version>9.5.0</version>
+</dependency>
+<!-- plus ONLY the reactive library you use, e.g.: -->
+<dependency>
+ <groupId>io.projectreactor</groupId>
+ <artifactId>reactor-core</artifactId>
+ <version>3.6.11</version>
+</dependency>
+```
+
#### Health Probe SPI + Resource (TODO-65)
`juneau-rest-server` now includes a built-in probe SPI and aggregation
resource under
diff --git a/pages/topics/10.20m.RestServerReactive.md
b/pages/topics/10.20m.RestServerReactive.md
new file mode 100644
index 0000000000..b5fd87afe7
--- /dev/null
+++ b/pages/topics/10.20m.RestServerReactive.md
@@ -0,0 +1,212 @@
+---
+title: "Reactive-Streams Returns"
+slug: RestServerReactive
+---
+
+`@RestOp` handler methods may return reactive-streams values. **The feature is
fully opt-in by adding a module to the classpath** — a bare
`juneau-rest-server` has zero reactive behavior and its `DefaultConfig` wires
no reactive processor.
+
+Two layers stack on top of core:
+
+| Classpath | Enables |
+|---|---|
+| bare `juneau-rest-server` | *(no reactive behavior — a `Flow.Publisher`
return is serialized as an ordinary object)* |
+| `+ juneau-rest-server-reactive` | JDK-native
`java.util.concurrent.Flow.Publisher<T>` returns (dependency-free) |
+| `+ juneau-rest-server-reactor` | Project Reactor (`Mono` / `Flux`), RxJava 3
(`Single` / `Maybe` / `Completable` / `Flowable` / `Observable`), and the
generic `org.reactivestreams.Publisher<T>` |
+
+Both modules are server-side only. `juneau-rest-server-reactive` is
dependency-free; `juneau-rest-server-reactor` declares Reactor / RxJava /
Reactive-Streams in `provided` scope.
+
+This feature shares the [async-returns](/docs/topics/RestServerAsyncDispatch)
plumbing: single-value reactive types collapse onto the existing
`CompletableFuture` path, and multi-value streams are rendered as a buffered
list, Server-Sent Events, or newline-delimited JSON.
+
+## Quick start
+
+```java
+@Rest(path="/feed")
+public class FeedResource extends BasicRestServlet {
+
+ @Inject ArticleService articles;
+
+ // Single value — uses the async (CompletableFuture) path under the hood.
+ @RestGet("/{id}")
+ public Mono<Article> get(@Path String id) {
+ return articles.findById(id);
+ }
+
+ // Bounded stream — collected into a JSON array (the default shape).
+ @RestGet("/recent")
+ public Flux<Article> recent() {
+ return articles.recent(20);
+ }
+
+ // Unbounded stream as Server-Sent Events.
+ @RestGet("/live")
+ public Flux<SseEvent> live(RestResponse res) {
+ res.setContentType("text/event-stream");
+ return articles.liveEvents();
+ }
+}
+```
+
+## Architecture
+
+A single response processor — `ReactiveResponseProcessor`
(`org.apache.juneau.rest.reactive`, shipped in the opt-in
`juneau-rest-server-reactive` module) — is the shared spine for all reactive
return-type support.
+
+**Auto-registration (no manual wiring).** The processor is *not* listed in
`DefaultConfig`. Instead, `juneau-rest-server-reactive` ships a
`META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor` provider
file naming `ReactiveResponseProcessor`. `RestContext` discovers
module-contributed response processors via `ServiceLoader` and front-loads them
ahead of `AsyncResponseProcessor` in the chain. On a bare `juneau-rest-server`
classpath this discovery finds nothing, so the ch [...]
+
+1. Natively recognizes `java.util.concurrent.Flow.Publisher<T>` (no external
dependency).
+2. For any other return value, consults registered `ReactiveStreamsAdapter`
providers (discovered via the same `ServiceLoader` mechanism). The
`juneau-rest-server-reactor` module supplies adapters for Reactor, RxJava 3,
and the generic Reactive-Streams `Publisher`.
+
+Each adapter converts its library's type into one of two JDK-native shapes,
expressed by the `Adaptation` value type:
+
+- a **single-value** `java.util.concurrent.CompletionStage` (e.g. `Mono`,
`Single`, `Maybe`, `Completable`), or
+- a **multi-value** `Flow.Publisher` (e.g. `Flux`, `Flowable`, `Observable`,
`Publisher`).
+
+Single-value adaptations are handed back to the response chain as a
`CompletionStage`, where the existing `AsyncResponseProcessor` picks them up —
so they inherit its timeout,
[`@Rest(asyncCompletionExecutor)`](/docs/topics/RestServerAsyncDispatch)
routing (TODO-118), and [SLF4J MDC
bridge](/docs/topics/RestServerAsyncDispatch) (TODO-117) for free.
+
+```mermaid
+flowchart TD
+ A["@RestOp returns reactive value"] --> B{ReactiveResponseProcessor}
+ B -->|Flow.Publisher| D[dispatch stream]
+ B -->|other type| C{ReactiveStreamsAdapter}
+ C -->|single| E[CompletionStage]
+ C -->|stream| D
+ E --> F[AsyncResponseProcessor: timeout, executor, MDC]
+ D --> G{shape?}
+ G -->|Buffer default| H[collect to List → JSON]
+ G -->|text/event-stream| I[SSE frames]
+ G -->|application/x-ndjson| J[NDJSON lines]
+```
+
+## Response shapes for multi-value streams
+
+A streaming publisher is rendered as one of three shapes, **selected by the
negotiated response media type** — the handler's `Content-Type` (set via
`res.setContentType(...)` or `@RestOp(produces=...)`), and failing that, the
request `Accept` header:
+
+| Shape | Trigger media type(s) | Behavior |
+|---|---|---|
+| **Buffer** (default) | anything else / unspecified | All elements collected
into a `List` and serialized through the normal serializer chain (e.g. a JSON
array). The collection is wrapped in a `CompletableFuture` and handed to the
async path, so a slow producer never blocks the request thread. |
+| **SSE** | `text/event-stream` | Each element emitted as a Server-Sent-Events
frame. `SseEvent` elements are written verbatim; any other element type is
JSON-encoded into the `data:` field. |
+| **NDJSON** | `application/x-ndjson`, `application/jsonl`,
`application/json-seq` | Each element JSON-encoded on its own line. |
+
+The default-to-buffer choice is deliberate: a bare `Flux<Article>` with no
content-type hint produces a normal JSON array, which is the least surprising
result for the majority of callers. Streaming is an explicit opt-in via the
content type.
+
+### Choosing the shape
+
+```java
+// Buffer (default) — JSON array.
+@RestGet("/page")
+public Flux<Row> page() { return repo.page(50); }
+
+// SSE — handler sets the content type.
+@RestGet("/events")
+public Flux<SseEvent> events(RestResponse res) {
+ res.setContentType("text/event-stream");
+ return bus.subscribe();
+}
+
+// NDJSON — via @RestOp(produces=...) instead of setting it imperatively.
+@RestGet(path="/export", produces="application/x-ndjson")
+public Flux<Row> export() { return repo.everything(); }
+
+// Shape can also be driven by the client's Accept header when the handler
+// leaves the content type unset.
+```
+
+## Backpressure
+
+Streaming subscribers request **one element at a time** — `request(1)` on
subscribe, and `request(1)` again after each frame is written and flushed.
Because writing to the servlet output stream blocks until the socket accepts
the bytes, this provides natural backpressure: the producer is paced by the
client's drain rate and the server-side buffer does not grow without bound.
+
+Buffer-shape subscribers request `Long.MAX_VALUE`, since the collection is
bounded by the publisher's own completion. Apply your own bound (`.take(n)`,
`.limitRequest(n)`, paging) before returning a publisher you intend to buffer —
an unbounded `Flux` buffered to a `List` will accumulate in memory.
+
+`Observable` (which has no native backpressure) is converted to a `Flowable`
with `BackpressureStrategy.BUFFER` before adaptation.
+
+## Threading, executors, and MDC
+
+- **Buffer-shape** responses route through `AsyncResponseProcessor` and
therefore honor `@Rest(asyncCompletionExecutor)` and the SLF4J MDC bridge
exactly as a `CompletableFuture` return would.
+- **Streaming-shape** frame writes happen on whichever thread the publisher
emits on (the Reactor / RxJava scheduler). This processor does **not** impose a
`subscribeOn(...)`, so it never fights the library's own scheduler model. When
MDC propagation is enabled (`RestContext.isMdcAsyncPropagation()`), the
request-thread MDC snapshot is reinstalled around each `onNext` / terminal
callback so log statements emitted while writing a frame see the request's
diagnostic context.
+
+## Synchronous fallback
+
+In environments where `HttpServletRequest.startAsync()` is unsupported —
notably Juneau's `MockServletRequest` used by `MockRestClient` in unit tests —
streaming subscribes synchronously and blocks the request thread until the
publisher terminates (bounded by the configured async timeout), writing frames
as they arrive. This keeps the unit-test surface working without a real servlet
container.
+
+## The modules
+
+### `juneau-rest-server-reactive` (JDK-native core)
+
+Dependency-free. Carries `ReactiveResponseProcessor`, the
`ReactiveStreamsAdapter` SPI, the `Adaptation` value type, and the
`META-INF/services/...ResponseProcessor` auto-registration file. Add this alone
to enable `Flow.Publisher<T>` returns with no third-party library:
+
+```xml
+<dependency>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest-server-reactive</artifactId>
+ <version>9.5.0</version>
+</dependency>
+```
+
+### `juneau-rest-server-reactor` (third-party bridge)
+
+The bridge module is **server-side only** and opt-in, and depends on
`juneau-rest-server-reactive`. All three backing libraries are
`provided`-scope: a `dependency:tree` on `juneau-rest-server` (or
`juneau-rest-server-reactive`) never surfaces `reactor-core`, `rxjava`, or
`reactive-streams`. Adapters are discovered lazily via `ServiceLoader` and
skipped at runtime if their backing library is absent from the classpath, so
you add the module plus **only the reactive library you actually us [...]
+
+```xml
+<dependency>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest-server-reactor</artifactId>
+ <version>9.5.0</version>
+</dependency>
+<dependency>
+ <groupId>io.projectreactor</groupId>
+ <artifactId>reactor-core</artifactId>
+ <version>3.6.11</version>
+</dependency>
+```
+
+| Adapter | Handles | Adapts to |
+|---|---|---|
+| `ReactorReactiveAdapter` | `Mono<T>` / `Flux<T>` | `Mono` →
`CompletionStage` (`Mono.toFuture()`); `Flux` → `Flow.Publisher`
(`FlowAdapters`) |
+| `RxJavaReactiveAdapter` | `Single` / `Maybe` / `Completable` / `Flowable` /
`Observable` | single types → `CompletionStage`; `Flowable` → `Flow.Publisher`;
`Observable` → `Flowable(BUFFER)` → `Flow.Publisher` |
+| `ReactiveStreamsPublisherAdapter` | generic
`org.reactivestreams.Publisher<T>` | `Flow.Publisher` (`FlowAdapters`);
registered last so `Mono`/`Flux` resolve as single/stream first |
+
+An empty `Mono` / `Maybe` / `Completable` completes the response with a `null`
body.
+
+### Extending with your own adapter
+
+Bridge any other reactive library by implementing `ReactiveStreamsAdapter` and
registering it via `META-INF/services`:
+
+```java
+public class MyLibAdapter implements ReactiveStreamsAdapter {
+ @Override public boolean canAdapt(Object v) { return v instanceof MyAsync
|| v instanceof MyStream; }
+ @Override public Adaptation adapt(Object v) {
+ if (v instanceof MyAsync<?> a) return
Adaptation.single(a.toCompletionStage());
+ return Adaptation.stream(FlowAdapters.toFlowPublisher(((MyStream<?>)
v).toReactiveStreams()));
+ }
+}
+```
+
+```
+# META-INF/services/org.apache.juneau.rest.reactive.ReactiveStreamsAdapter
+com.example.MyLibAdapter
+```
+
+## Configuration summary
+
+| Setting | Where | Effect |
+|---|---|---|
+| return `Flow.Publisher<T>` | `@RestOp` method (with
`juneau-rest-server-reactive`) | JDK-native streaming, no third-party
dependency |
+| return `Mono` / `Single` / `Maybe` / `Completable` | `@RestOp` method (with
`juneau-rest-server-reactor`) | single value via async path |
+| return `Flux` / `Flowable` / `Observable` / `Publisher` | `@RestOp` method
(with `juneau-rest-server-reactor`) | stream — shape by content type |
+| `res.setContentType("text/event-stream")` | handler body | force SSE shape |
+| `res.setContentType("application/x-ndjson")` | handler body | force NDJSON
shape |
+| (no content type) | — | buffer to list (JSON array) |
+
+## Acceptance verification
+
+The opt-in contract is verified from both sides:
+
+| Behavior | Test | Module |
+|---|---|---|
+| **Bare** `juneau-rest-server` does NOT process `Flow.Publisher` reactively |
`ReactiveOptIn_BareServer_Test` | `juneau-rest-mock` (no reactive module on its
classpath) |
+| `Flow.Publisher` buffered to JSON array (default) |
`ReactiveResponseProcessor_Test` | `juneau-utest` |
+| `Flow.Publisher<SseEvent>` → SSE frames | `ReactiveResponseProcessor_Test` |
`juneau-utest` |
+| POJO element stream → SSE / NDJSON | `ReactiveResponseProcessor_Test` |
`juneau-utest` |
+| Publisher error propagation | `ReactiveResponseProcessor_Test` |
`juneau-utest` |
+| Reactor `Mono` / `Flux` adaptation | `ReactiveBridge_Test` | `juneau-utest` |
+| RxJava `Single` / `Maybe` / `Completable` / `Flowable` / `Observable`
adaptation | `ReactiveBridge_Test` | `juneau-utest` |
+| Generic `org.reactivestreams.Publisher` adaptation | `ReactiveBridge_Test` |
`juneau-utest` |
diff --git a/sidebars.ts b/sidebars.ts
index e0110a361a..5a4c61afbd 100644
--- a/sidebars.ts
+++ b/sidebars.ts
@@ -1571,6 +1571,11 @@ const sidebars: SidebarsConfig = {
id:
'topics/10.20h.RestServerAsyncDispatch',
label: '10.20h. Async
Returns + Virtual-Thread Dispatch',
},
+ {
+ type: 'doc',
+ id:
'topics/10.20m.RestServerReactive',
+ label: '10.20m.
Reactive-Streams Returns',
+ },
{
type: 'doc',
id:
'topics/10.21.BuiltInParameters',
