You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@juneau.apache.org by ja...@apache.org on 2026/05/29 16:31:08 UTC
(juneau) branch master updated: 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
This is an automated email from the ASF dual-hosted git repository.
jamesbognar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/juneau.git
The following commit(s) were added to refs/heads/master by this push:
new 4369f411e8 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
4369f411e8 is described below
commit 4369f411e8f91007db39db3be8b3d62a4f99dd49
Author: James Bognar <ja...@apache.org>
AuthorDate: Fri May 29 12:31:02 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
---
juneau-rest/juneau-rest-mock/pom.xml | 11 +
.../reactive/ReactiveOptIn_BareServer_Test.java | 83 ++++
.../pom.xml | 74 ++--
.../apache/juneau/rest/reactive/Adaptation.java | 95 +++++
.../rest/reactive/ReactiveResponseProcessor.java | 442 +++++++++++++++++++++
.../rest/reactive/ReactiveStreamsAdapter.java | 78 ++++
.../apache/juneau/rest/reactive/package-info.java | 54 +++
....apache.juneau.rest.processor.ResponseProcessor | 1 +
juneau-rest/juneau-rest-server-reactor/pom.xml | 140 +++++++
.../bridge/ReactiveStreamsPublisherAdapter.java | 47 +++
.../reactive/bridge/ReactorReactiveAdapter.java | 54 +++
.../reactive/bridge/RxJavaReactiveAdapter.java | 64 +++
.../juneau/rest/reactive/bridge/package-info.java | 46 +++
...che.juneau.rest.reactive.ReactiveStreamsAdapter | 3 +
.../java/org/apache/juneau/rest/RestContext.java | 59 +++
juneau-rest/pom.xml | 2 +
juneau-utest/pom.xml | 49 +++
.../reactive/ReactiveResponseProcessor_Test.java | 251 ++++++++++++
.../rest/reactive/bridge/ReactiveBridge_Test.java | 191 +++++++++
juneau-utest/test-run-history.tsv | 1 +
20 files changed, 1716 insertions(+), 29 deletions(-)
diff --git a/juneau-rest/juneau-rest-mock/pom.xml b/juneau-rest/juneau-rest-mock/pom.xml
index 838edfe777..ab51c1f18c 100644
--- a/juneau-rest/juneau-rest-mock/pom.xml
+++ b/juneau-rest/juneau-rest-mock/pom.xml
@@ -50,6 +50,17 @@
<artifactId>juneau-rest-client-classic</artifactId>
<version>${project.version}</version>
</dependency>
+ <!--
+ Test-only: the "bare server has zero reactive behavior" guard (ReactiveOptIn_BareServer_Test).
+ This module's test classpath intentionally has NO reactive module, so it is the only place the
+ opt-in contract can be asserted from the absence side. See FINISHED-119/120.
+ -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/juneau-rest/juneau-rest-mock/src/test/java/org/apache/juneau/rest/reactive/ReactiveOptIn_BareServer_Test.java b/juneau-rest/juneau-rest-mock/src/test/java/org/apache/juneau/rest/reactive/ReactiveOptIn_BareServer_Test.java
new file mode 100644
index 0000000000..89a4c0ba2d
--- /dev/null
+++ b/juneau-rest/juneau-rest-mock/src/test/java/org/apache/juneau/rest/reactive/ReactiveOptIn_BareServer_Test.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.concurrent.*;
+
+import org.apache.juneau.json.*;
+import org.apache.juneau.rest.annotation.*;
+import org.apache.juneau.rest.mock.classic.*;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that a <b>bare</b> {@code juneau-rest-server} (no {@code juneau-rest-server-reactive} /
+ * {@code juneau-rest-server-reactor} module on the classpath) has <b>zero</b> reactive behavior.
+ *
+ * <p>
+ * This module's test classpath intentionally has no reactive module, so {@code RestContext}'s
+ * {@link java.util.ServiceLoader}-based response-processor discovery finds no provider and
+ * {@code ReactiveResponseProcessor} is never added to the chain. A {@code @RestOp} returning a
+ * {@link java.util.concurrent.Flow.Publisher Flow.Publisher} is therefore treated as an ordinary POJO
+ * return value and serialized through the normal serializer chain — it is NOT collected into a
+ * JSON array (the reactive buffer shape) and NOT streamed.
+ *
+ * <p>
+ * The positive activation tests live in {@code juneau-utest} (where the reactive module IS on the
+ * classpath); this negative test cannot live there because the module's presence auto-activates the
+ * processor for every resource. See {@code FINISHED-119}/{@code FINISHED-120} for the opt-in design.
+ */
+class ReactiveOptIn_BareServer_Test {
+
+ public static final class Pojo {
+ public final String name;
+ public final int value;
+ public Pojo(String name, int value) { this.name = name; this.value = value; }
+ }
+
+ /** Minimal JDK Flow.Publisher; on a bare server it is never subscribed to (no reactive processor). */
+ static final class ListPublisher<T> implements Flow.Publisher<T> {
+ @Override public void subscribe(Flow.Subscriber<? super T> sub) {
+ sub.onSubscribe(new Flow.Subscription() {
+ @Override public void request(long n) { /* no-op */ }
+ @Override public void cancel() { /* no-op */ }
+ });
+ }
+ }
+
+ @Rest(serializers = JsonSerializer.class)
+ public static class A {
+ @RestGet("/flux")
+ public Flow.Publisher<Pojo> flux() {
+ return new ListPublisher<>();
+ }
+ }
+
+ private static final MockRestClient CA = MockRestClient.buildLax(A.class);
+
+ @Test void a01_bareServer_doesNotProcessFlowPublisherReactively() throws Exception {
+ var content = CA.get("/flux").accept("application/json").run().getContent().asString();
+ // Reactive (buffer-shape) handling would have subscribed to the publisher and produced a JSON
+ // array of the emitted Pojo elements. With no reactive module on the classpath the publisher is
+ // instead serialized as a plain (property-less) bean — so the element payload must be absent.
+ assertFalse(content.contains("\"name\""),
+ "Bare juneau-rest-server must not reactively process Flow.Publisher returns. Body was: " + content);
+ assertFalse(content.startsWith("["),
+ "Bare juneau-rest-server must not buffer a Flow.Publisher into a JSON array. Body was: " + content);
+ }
+}
diff --git a/juneau-rest/juneau-rest-mock/pom.xml b/juneau-rest/juneau-rest-server-reactive/pom.xml
similarity index 63%
copy from juneau-rest/juneau-rest-mock/pom.xml
copy to juneau-rest/juneau-rest-server-reactive/pom.xml
index 838edfe777..3ed903b345 100644
--- a/juneau-rest/juneau-rest-mock/pom.xml
+++ b/juneau-rest/juneau-rest-server-reactive/pom.xml
@@ -25,9 +25,9 @@
<version>9.5.0-SNAPSHOT</version>
</parent>
- <artifactId>juneau-rest-mock</artifactId>
- <name>Apache Juneau REST Mock</name>
- <description>Apache Juneau REST mock API</description>
+ <artifactId>juneau-rest-server-reactive</artifactId>
+ <name>Apache Juneau REST Server Reactive Streams Core</name>
+ <description>Apache Juneau REST Server - JDK-native Flow.Publisher return-type support + ReactiveStreamsAdapter SPI (opt-in, dependency-free)</description>
<packaging>bundle</packaging>
<properties>
@@ -35,25 +35,35 @@
</properties>
<dependencies>
+ <!--
+ Dependency-free by design. This module carries only the JDK-native Flow.Publisher
+ response processor, the ReactiveStreamsAdapter SPI, and the Adaptation value type, plus
+ the META-INF/services/...ResponseProcessor file that auto-registers the processor when
+ this jar is on the classpath. No third-party reactive library is referenced — the
+ Reactor / RxJava / Reactive-Streams adapters live in juneau-rest-server-reactor.
+ -->
<dependency>
<groupId>org.apache.juneau</groupId>
<artifactId>juneau-rest-server</artifactId>
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.juneau</groupId>
- <artifactId>juneau-rest-client</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.juneau</groupId>
- <artifactId>juneau-rest-client-classic</artifactId>
- <version>${project.version}</version>
- </dependency>
</dependencies>
<build>
<plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-source-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <phase>verify</phase>
+ <goals>
+ <goal>jar-no-fork</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
@@ -61,25 +71,12 @@
<configuration>
<supportIncrementalBuild>true</supportIncrementalBuild>
</configuration>
- <executions>
- <execution>
- <id>bundle-manifest</id>
- <phase>process-classes</phase>
- <goals>
- <goal>manifest</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-source-plugin</artifactId>
<executions>
<execution>
- <id>attach-sources</id>
- <phase>verify</phase>
+ <id>bundle-manifest</id>
+ <phase>process-classes</phase>
<goals>
- <goal>jar-no-fork</goal>
+ <goal>manifest</goal>
</goals>
</execution>
</executions>
@@ -88,6 +85,25 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
</plugin>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>default-prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>default-report</id>
+ <phase>prepare-package</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
diff --git a/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/Adaptation.java b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/Adaptation.java
new file mode 100644
index 0000000000..5042f9e7a0
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/Adaptation.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive;
+
+import static org.apache.juneau.commons.utils.AssertionUtils.*;
+
+import java.util.concurrent.*;
+
+/**
+ * The result of a {@link ReactiveStreamsAdapter#adapt(Object) ReactiveStreamsAdapter.adapt(...)} call:
+ * either a single-value {@link CompletionStage} or a multi-value
+ * {@link java.util.concurrent.Flow.Publisher Flow.Publisher}.
+ *
+ * <p>
+ * Exactly one of the two shapes is present. Single-value adaptations (e.g. Reactor {@code Mono},
+ * RxJava {@code Single} / {@code Maybe}) collapse onto the existing {@code CompletableFuture} async
+ * response path. Streaming adaptations (e.g. Reactor {@code Flux}, RxJava {@code Flowable} /
+ * {@code Observable}, a Reactive-Streams {@code Publisher}) are subscribed to and rendered as a
+ * buffered list, an SSE stream, or an NDJSON stream depending on the negotiated response media type.
+ *
+ * @see ReactiveStreamsAdapter
+ * @see ReactiveResponseProcessor
+ * @since 9.5.0
+ */
+public final class Adaptation {
+
+ private final CompletionStage<?> single;
+ private final java.util.concurrent.Flow.Publisher<?> stream;
+
+ private Adaptation(CompletionStage<?> single, java.util.concurrent.Flow.Publisher<?> stream) {
+ this.single = single;
+ this.stream = stream;
+ }
+
+ /**
+ * Creates a single-value adaptation backed by a {@link CompletionStage}.
+ *
+ * @param value The completion stage. Must not be {@code null}.
+ * @return A new adaptation.
+ */
+ public static Adaptation single(CompletionStage<?> value) {
+ return new Adaptation(assertArgNotNull("value", value), null);
+ }
+
+ /**
+ * Creates a streaming adaptation backed by a {@link java.util.concurrent.Flow.Publisher Flow.Publisher}.
+ *
+ * @param value The publisher. Must not be {@code null}.
+ * @return A new adaptation.
+ */
+ public static Adaptation stream(java.util.concurrent.Flow.Publisher<?> value) {
+ return new Adaptation(null, assertArgNotNull("value", value));
+ }
+
+ /**
+ * Returns whether this adaptation is a multi-value stream (vs. a single value).
+ *
+ * @return {@code true} if this is a {@link java.util.concurrent.Flow.Publisher Flow.Publisher} adaptation.
+ */
+ public boolean isStream() {
+ return stream != null;
+ }
+
+ /**
+ * Returns the single-value completion stage, or {@code null} if this is a streaming adaptation.
+ *
+ * @return The completion stage, or {@code null}.
+ */
+ public CompletionStage<?> single() {
+ return single;
+ }
+
+ /**
+ * Returns the streaming publisher, or {@code null} if this is a single-value adaptation.
+ *
+ * @return The publisher, or {@code null}.
+ */
+ public java.util.concurrent.Flow.Publisher<?> stream() {
+ return stream;
+ }
+}
diff --git a/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/ReactiveResponseProcessor.java b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/ReactiveResponseProcessor.java
new file mode 100644
index 0000000000..4fae418237
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/ReactiveResponseProcessor.java
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive;
+
+import static jakarta.servlet.http.HttpServletResponse.*;
+
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.*;
+import java.util.function.*;
+import java.util.logging.*;
+
+import jakarta.servlet.*;
+import jakarta.servlet.http.*;
+
+import org.apache.juneau.http.response.*;
+import org.apache.juneau.marshaller.*;
+import org.apache.juneau.rest.*;
+import org.apache.juneau.rest.processor.*;
+import org.apache.juneau.rest.util.*;
+import org.apache.juneau.serializer.*;
+import org.apache.juneau.sse.*;
+
+/**
+ * Response processor that bridges <a class="doclink" href="https://www.reactive-streams.org/">Reactive Streams</a>
+ * return values from {@code @RestOp} handlers to Juneau's response pipeline.
+ *
+ * <h5 class='topic'>Activation (opt-in)</h5>
+ * <p>
+ * This processor ships in the opt-in {@code juneau-rest-server-reactive} module and is <b>not</b> wired
+ * into {@code DefaultConfig}. A bare {@code juneau-rest-server} has zero reactive behavior. When the
+ * {@code juneau-rest-server-reactive} jar is on the classpath, its
+ * {@code META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor} provider file is discovered
+ * by {@code RestContext} via {@link java.util.ServiceLoader} and this processor is front-loaded ahead of
+ * {@code AsyncResponseProcessor} in the chain — no {@code @Rest(responseProcessors=...)} entry required.
+ *
+ * <h5 class='topic'>What it handles</h5>
+ * <p>
+ * This is the single, shared spine for all reactive return-type support. It natively understands the
+ * JDK type {@link java.util.concurrent.Flow.Publisher Flow.Publisher<T>} (no external dependency
+ * required) and, through registered {@link ReactiveStreamsAdapter} providers, any third-party reactive
+ * type the opt-in {@code juneau-rest-server-reactor} module adapts to it (Project Reactor
+ * {@code Mono} / {@code Flux}, RxJava 3 {@code Single} / {@code Maybe} / {@code Flowable} /
+ * {@code Observable}, and the Reactive-Streams {@code org.reactivestreams.Publisher}).
+ *
+ * <p>
+ * Single-value reactive types (e.g. {@code Mono}, {@code Single}) are adapted to a {@link CompletionStage}
+ * and collapse onto the existing {@link AsyncResponseProcessor} async path, inheriting its timeout,
+ * completion-executor ({@code @Rest(asyncCompletionExecutor)}), and MDC bridging behavior for free.
+ *
+ * <h5 class='topic'>Response shapes for multi-value streams</h5>
+ * <p>
+ * A streaming publisher is rendered as one of three shapes, selected by the negotiated response media
+ * type (the handler's {@link RestResponse#setContentType(String) Content-Type}, then the request
+ * {@code Accept} header):
+ * <ul>
+ * <li><b>SSE</b> ({@code text/event-stream}) — each element is emitted as a Server-Sent-Events
+ * frame. {@link SseEvent} elements are written verbatim; any other element type is JSON-encoded
+ * into the {@code data:} field.
+ * <li><b>NDJSON</b> ({@code application/x-ndjson}, {@code application/jsonl}) — each element is
+ * JSON-encoded on its own line.
+ * <li><b>Buffer</b> (default, any other media type) — all elements are collected into a
+ * {@link java.util.List List} and serialized through the normal serializer chain (e.g. a JSON
+ * array). The collection is wrapped in a {@link CompletableFuture} and handed to the async path,
+ * so a slow producer never blocks the request thread.
+ * </ul>
+ *
+ * <h5 class='topic'>Backpressure</h5>
+ * <p>
+ * Streaming subscribers request one element at a time ({@code request(1)} on subscribe and again after
+ * each frame is written and flushed). Because writing to the servlet output stream blocks until the
+ * socket accepts the bytes, this gives 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 {@link Long#MAX_VALUE} since the collection is bounded by the publisher's own completion.
+ *
+ * <h5 class='topic'>Threading, executors, and MDC</h5>
+ * <p>
+ * Buffer-shape responses route through {@link AsyncResponseProcessor} and therefore honor
+ * {@code @Rest(asyncCompletionExecutor)} (TODO-118) and the SLF4J MDC bridge (TODO-117). Streaming-shape
+ * frame writes happen on whichever thread the publisher emits on (the Reactor / RxJava scheduler); this
+ * processor does not impose a {@code subscribeOn(...)} so it never fights the library's scheduler model.
+ * When MDC propagation is enabled, the request-thread MDC snapshot is reinstalled around each
+ * {@code onNext} / terminal callback via {@link MdcAsyncListener} so log statements emitted while writing
+ * a frame see the request's diagnostic context.
+ *
+ * <h5 class='topic'>Synchronous fallback</h5>
+ * <p>
+ * In environments where {@link HttpServletRequest#startAsync()} is unsupported (notably Juneau's
+ * {@code MockServletRequest}), 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.
+ *
+ * @see ReactiveStreamsAdapter
+ * @see AsyncResponseProcessor
+ * @since 9.5.0
+ */
+public class ReactiveResponseProcessor implements ResponseProcessor {
+
+ private static final Logger LOG = Logger.getLogger(ReactiveResponseProcessor.class.getName());
+
+ private static volatile List<ReactiveStreamsAdapter> adapters;
+
+ private enum Shape { BUFFER, SSE, NDJSON }
+
+ @FunctionalInterface
+ private interface FrameEncoder {
+ void write(FinishablePrintWriter w, Object element) throws IOException, SerializeException;
+ }
+
+ @Override /* Overridden from ResponseProcessor */
+ public int process(RestOpSession opSession) throws IOException, BasicHttpException {
+ var res = opSession.getResponse();
+ var content = res.getContent().orElse(null);
+
+ if (content == null)
+ return NEXT;
+
+ if (content instanceof Flow.Publisher<?> pub)
+ return dispatch(opSession, pub);
+
+ var a = adapters();
+ for (var adapter : a) {
+ if (adapter.canAdapt(content)) {
+ var adaptation = adapter.adapt(content);
+ if (adaptation.isStream())
+ return dispatch(opSession, adaptation.stream());
+ res.setContent(adaptation.single());
+ return RESTART;
+ }
+ }
+
+ return NEXT;
+ }
+
+ private int dispatch(RestOpSession opSession, Flow.Publisher<?> pub) throws IOException {
+ var shape = resolveShape(opSession);
+ if (shape == Shape.BUFFER)
+ return handleBuffer(opSession, pub);
+ return handleStream(opSession, pub, shape);
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // Buffer shape — collect into a List, hand to the async path.
+ // -----------------------------------------------------------------------------------------------------------------
+
+ private int handleBuffer(RestOpSession opSession, Flow.Publisher<?> pub) {
+ var cf = new CompletableFuture<List<Object>>();
+ pub.subscribe(new CollectingSubscriber(cf));
+ opSession.getResponse().setContent(cf);
+ return RESTART;
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // Streaming shapes — SSE / NDJSON.
+ // -----------------------------------------------------------------------------------------------------------------
+
+ @SuppressWarnings({
+ "java:S3776", // Streaming setup is branchy by nature (content-type prep + async-vs-sync fallback).
+ "java:S1141" // Nested try/catch cleanly separates startAsync recovery from the sync fallback.
+ })
+ private int handleStream(RestOpSession opSession, Flow.Publisher<?> pub, Shape shape) throws IOException {
+ var res = opSession.getResponse();
+ prepareStreamingHeaders(res, shape);
+
+ var writer = res.getNegotiatedWriter();
+ FrameEncoder encoder = shape == Shape.SSE
+ ? ReactiveResponseProcessor::writeSseFrame
+ : ReactiveResponseProcessor::writeNdjsonFrame;
+
+ var req = opSession.getRequest().getHttpServletRequest();
+ var mdc = opSession.getRestContext().isMdcAsyncPropagation()
+ ? MdcAsyncListener.snapshot()
+ : null;
+
+ AsyncContext asyncCtx = null;
+ if (req.isAsyncSupported()) {
+ try {
+ asyncCtx = req.startAsync();
+ } catch (IllegalStateException e) {
+ asyncCtx = null; // Already committed, or async not actually supported.
+ }
+ }
+
+ if (asyncCtx != null) {
+ req.setAttribute(AsyncResponseProcessor.ATTR_ASYNC_DISPATCH_OWNED, Boolean.TRUE);
+ var ac = asyncCtx;
+ ac.setTimeout(0); // No artificial timeout — streams are paced by the producer / client disconnect.
+ pub.subscribe(new StreamingSubscriber(res, writer, encoder, mdc, t -> completeAsync(ac, res, t)));
+ return FINISHED;
+ }
+
+ // Synchronous fallback (MockServletRequest et al.): block until the publisher terminates.
+ var done = new CompletableFuture<Void>();
+ pub.subscribe(new StreamingSubscriber(res, writer, encoder, mdc,
+ t -> { if (t == null) done.complete(null); else done.completeExceptionally(t); }));
+ awaitSync(done, syncTimeoutMillis(opSession));
+ try {
+ writer.flush();
+ res.flushBuffer();
+ } catch (IOException e) {
+ LOG.log(Level.FINEST, e, () -> "Final flush of synchronous reactive stream failed: " + e.getMessage());
+ }
+ return FINISHED;
+ }
+
+ private static void prepareStreamingHeaders(RestResponse res, Shape shape) {
+ var ct = res.getContentType();
+ if (shape == Shape.SSE) {
+ if (ct == null || ! ct.contains("event-stream"))
+ res.setContentType(SseSerializer.MEDIA_TYPE);
+ res.setHeader("X-Content-Type-Options", "nosniff");
+ } else if (ct == null) {
+ res.setContentType("application/x-ndjson");
+ }
+ res.setHeader("Cache-Control", "no-cache");
+ res.setHeader("Content-Encoding", "identity");
+ }
+
+ @SuppressWarnings("java:S2142") // Interrupt flag restored before returning.
+ private static void awaitSync(CompletableFuture<Void> done, long timeoutMs) {
+ try {
+ done.get(timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (TimeoutException | ExecutionException e) {
+ LOG.log(Level.FINE, e, () -> "Synchronous reactive stream did not complete cleanly: " + e.getMessage());
+ }
+ }
+
+ private static void completeAsync(AsyncContext asyncCtx, RestResponse res, Throwable error) {
+ try {
+ if (error != null)
+ LOG.log(Level.FINE, error, () -> "Reactive stream terminated with error: " + error.getMessage());
+ if (! res.getHttpServletResponse().isCommitted())
+ res.flushBuffer();
+ } catch (IOException e) {
+ LOG.log(Level.FINEST, e, () -> "Flush during async stream completion failed: " + e.getMessage());
+ } finally {
+ try {
+ asyncCtx.complete();
+ } catch (IllegalStateException e) {
+ LOG.log(Level.FINEST, e, () -> "AsyncContext.complete() raced with the container: " + e.getMessage());
+ }
+ }
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // Frame encoders.
+ // -----------------------------------------------------------------------------------------------------------------
+
+ @SuppressWarnings({ "java:S2095", "resource" }) // Writer is response-owned; we never close it.
+ private static void writeSseFrame(FinishablePrintWriter w, Object element) throws IOException, SerializeException {
+ if (element == null)
+ return;
+ if (element instanceof SseEvent ev) {
+ SseSerializer.DEFAULT.serialize(ev, w);
+ return;
+ }
+ var data = element instanceof CharSequence c ? c.toString() : Json.of(element);
+ SseSerializer.DEFAULT.serialize(new SseEvent(null, data), w);
+ }
+
+ @SuppressWarnings({ "java:S2095", "resource" }) // Writer is response-owned; we never close it.
+ private static void writeNdjsonFrame(FinishablePrintWriter w, Object element) throws IOException, SerializeException {
+ if (element == null)
+ return;
+ var json = element instanceof CharSequence c ? c.toString() : Json.of(element);
+ w.write(json);
+ w.write("\n");
+ w.flush();
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // Shape / timeout resolution.
+ // -----------------------------------------------------------------------------------------------------------------
+
+ private static Shape resolveShape(RestOpSession opSession) {
+ var res = opSession.getResponse();
+ var ct = res.getContentType();
+ var accept = opSession.getRequest().getHttpServletRequest().getHeader("Accept");
+ var probe = ((ct == null ? "" : ct) + "," + (accept == null ? "" : accept)).toLowerCase(Locale.ROOT);
+ if (probe.contains("event-stream"))
+ return Shape.SSE;
+ if (probe.contains("ndjson") || probe.contains("jsonl") || probe.contains("json-seq"))
+ return Shape.NDJSON;
+ return Shape.BUFFER;
+ }
+
+ private static long syncTimeoutMillis(RestOpSession opSession) {
+ var t = opSession.getContext().getAsyncTimeoutMillis();
+ if (t <= 0)
+ t = opSession.getRestContext().getAsyncTimeoutMillis();
+ return t > 0 ? t : AsyncResponseProcessor.DEFAULT_ASYNC_TIMEOUT_MILLIS;
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // Adapter discovery (ServiceLoader, cached, defensive against absent backing libraries).
+ // -----------------------------------------------------------------------------------------------------------------
+
+ private static List<ReactiveStreamsAdapter> adapters() {
+ var a = adapters;
+ if (a == null) {
+ synchronized (ReactiveResponseProcessor.class) {
+ a = adapters;
+ if (a == null) {
+ a = loadAdapters();
+ adapters = a;
+ }
+ }
+ }
+ return a;
+ }
+
+ private static List<ReactiveStreamsAdapter> loadAdapters() {
+ var out = new ArrayList<ReactiveStreamsAdapter>();
+ try {
+ var it = ServiceLoader.load(ReactiveStreamsAdapter.class).iterator();
+ while (it.hasNext()) {
+ try {
+ out.add(it.next());
+ } catch (ServiceConfigurationError | RuntimeException e) {
+ LOG.log(Level.FINE, e, () -> "Skipping reactive adapter (backing library likely absent): " + e.getMessage());
+ }
+ }
+ } catch (ServiceConfigurationError | RuntimeException e) {
+ LOG.log(Level.FINE, e, () -> "ServiceLoader for ReactiveStreamsAdapter failed: " + e.getMessage());
+ }
+ return List.copyOf(out);
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // Subscribers.
+ // -----------------------------------------------------------------------------------------------------------------
+
+ /** Collects all emitted elements into a list and completes the supplied future (buffer shape). */
+ private static final class CollectingSubscriber implements Flow.Subscriber<Object> {
+ private final CompletableFuture<List<Object>> cf;
+ private final List<Object> items = new ArrayList<>();
+
+ CollectingSubscriber(CompletableFuture<List<Object>> cf) {
+ this.cf = cf;
+ }
+
+ @Override public void onSubscribe(Flow.Subscription s) { s.request(Long.MAX_VALUE); }
+ @Override public void onNext(Object item) { items.add(item); }
+ @Override public void onError(Throwable t) { cf.completeExceptionally(t); }
+ @Override public void onComplete() { cf.complete(items); }
+ }
+
+ /** Writes each emitted element as a wire frame (SSE / NDJSON), one element at a time (bounded backpressure). */
+ private static final class StreamingSubscriber implements Flow.Subscriber<Object> {
+ private final RestResponse res;
+ private final FinishablePrintWriter writer;
+ private final FrameEncoder encoder;
+ private final Map<String,String> mdc;
+ private final Consumer<Throwable> onTerminate;
+ private final AtomicBoolean terminated = new AtomicBoolean();
+ private volatile boolean wrote;
+ private Flow.Subscription subscription;
+
+ StreamingSubscriber(RestResponse res, FinishablePrintWriter writer, FrameEncoder encoder,
+ Map<String,String> mdc, Consumer<Throwable> onTerminate) {
+ this.res = res;
+ this.writer = writer;
+ this.encoder = encoder;
+ this.mdc = mdc;
+ this.onTerminate = onTerminate;
+ }
+
+ @Override public void onSubscribe(Flow.Subscription s) {
+ subscription = s;
+ s.request(1);
+ }
+
+ @Override public void onNext(Object item) {
+ try {
+ withMdc(() -> writeFrame(item));
+ subscription.request(1);
+ } catch (RuntimeException e) {
+ subscription.cancel();
+ terminate(unwrap(e));
+ }
+ }
+
+ @Override public void onError(Throwable t) { terminate(t); }
+ @Override public void onComplete() { terminate(null); }
+
+ private void writeFrame(Object item) {
+ try {
+ encoder.write(writer, item);
+ writer.flush();
+ res.flushBuffer();
+ wrote = true;
+ } catch (IOException | SerializeException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void withMdc(Runnable r) {
+ MdcAsyncListener.wrap((BiConsumer<Object,Throwable>) (v, e) -> r.run(), mdc).accept(null, null);
+ }
+
+ private void terminate(Throwable t) {
+ if (! terminated.compareAndSet(false, true))
+ return;
+ withMdc(() -> {
+ if (t != null && ! wrote) {
+ try {
+ if (! res.getHttpServletResponse().isCommitted())
+ res.getHttpServletResponse().sendError(SC_INTERNAL_SERVER_ERROR);
+ } catch (IOException e) {
+ LOG.log(Level.FINEST, e, () -> "sendError on pre-write stream failure failed: " + e.getMessage());
+ }
+ }
+ });
+ onTerminate.accept(t);
+ }
+ }
+
+ private static Throwable unwrap(Throwable t) {
+ if ((t instanceof IllegalStateException || t instanceof CompletionException) && t.getCause() != null)
+ return t.getCause();
+ return t;
+ }
+}
diff --git a/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/ReactiveStreamsAdapter.java b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/ReactiveStreamsAdapter.java
new file mode 100644
index 0000000000..d46661961d
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/ReactiveStreamsAdapter.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive;
+
+import java.util.concurrent.*;
+
+/**
+ * SPI that bridges a third-party reactive return type to one of the two JDK-native shapes that the
+ * {@link ReactiveResponseProcessor} understands: a {@link CompletionStage} (single value) or a
+ * {@link java.util.concurrent.Flow.Publisher Flow.Publisher} (a stream of values).
+ *
+ * <h5 class='topic'>Why this exists</h5>
+ * <p>
+ * The opt-in {@code juneau-rest-server-reactive} module handles {@link java.util.concurrent.Flow.Publisher Flow.Publisher}
+ * (and, via the existing {@code AsyncResponseProcessor}, {@link CompletionStage}) using only JDK APIs —
+ * no external dependency. Libraries such as Project Reactor ({@code Mono} / {@code Flux}), RxJava 3
+ * ({@code Single} / {@code Maybe} / {@code Flowable} / {@code Observable}), and the
+ * Reactive-Streams {@code org.reactivestreams.Publisher} are NOT on the core classpath. The opt-in
+ * {@code juneau-rest-server-reactor} module ships {@code ReactiveStreamsAdapter} implementations that
+ * convert those types into the JDK-native shapes, so the streaming / buffering / SSE / NDJSON plumbing
+ * is implemented exactly once in the core.
+ *
+ * <h5 class='topic'>Registration</h5>
+ * <p>
+ * Adapters are discovered via the JDK {@link java.util.ServiceLoader} mechanism. A bridge module (or a
+ * consumer who wants a custom adapter) ships a
+ * {@code META-INF/services/org.apache.juneau.rest.reactive.ReactiveStreamsAdapter} file listing the
+ * implementation class names. The {@link ReactiveResponseProcessor} loads the providers once (lazily,
+ * at first use) and skips any provider whose backing library is absent from the runtime classpath
+ * (a {@link NoClassDefFoundError} / {@link ServiceConfigurationError} on instantiation is swallowed).
+ * This means a single {@code juneau-rest-server-reactor} jar can declare Reactor, RxJava, and
+ * Reactive-Streams adapters while the consumer pulls only the {@code provided}-scope libraries they
+ * actually want.
+ *
+ * @see ReactiveResponseProcessor
+ * @see Adaptation
+ * @since 9.5.0
+ */
+public interface ReactiveStreamsAdapter {
+
+ /**
+ * Returns whether this adapter recognizes the supplied return value.
+ *
+ * <p>
+ * Implementations must return {@code false} (rather than throw) for values they do not handle, and
+ * must be cheap to call — this is invoked on the response hot path for every non-null handler
+ * return value once at least one adapter is registered.
+ *
+ * @param value The {@code @RestOp} handler return value. Never {@code null}.
+ * @return {@code true} if {@link #adapt(Object)} can convert this value.
+ */
+ boolean canAdapt(Object value);
+
+ /**
+ * Converts the supplied reactive value to a JDK-native {@link Adaptation}.
+ *
+ * <p>
+ * Only called when {@link #canAdapt(Object)} returned {@code true} for the same value.
+ *
+ * @param value The reactive value to adapt. Never {@code null}.
+ * @return The adapted single-value or streaming shape. Never {@code null}.
+ */
+ Adaptation adapt(Object value);
+}
diff --git a/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/package-info.java b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/package-info.java
new file mode 100644
index 0000000000..1a37b4fd15
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactive/src/main/java/org/apache/juneau/rest/reactive/package-info.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Reactive-Streams return-type support for {@code @RestOp} handlers (opt-in, dependency-free).
+ *
+ * <p>
+ * This package is the JDK-native spine that lets {@code @RestOp} handlers return reactive values. It
+ * ships in the opt-in {@code juneau-rest-server-reactive} module — a bare {@code juneau-rest-server}
+ * has <b>no</b> reactive behavior and {@code DefaultConfig} does not wire any reactive processor. Merely
+ * placing this module on the classpath auto-registers {@link org.apache.juneau.rest.reactive.ReactiveResponseProcessor}
+ * via a {@code META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor} provider file, which
+ * {@code RestContext} discovers through {@link java.util.ServiceLoader} and front-loads ahead of
+ * {@code AsyncResponseProcessor}.
+ *
+ * <p>
+ * The processor handles {@link java.util.concurrent.Flow.Publisher Flow.Publisher<T>} directly (no
+ * external dependency) and exposes the {@link org.apache.juneau.rest.reactive.ReactiveStreamsAdapter} SPI
+ * so the further opt-in {@code juneau-rest-server-reactor} module can plug in Project Reactor, RxJava 3,
+ * and Reactive-Streams {@code Publisher} support without duplicating any of the streaming / buffering /
+ * SSE / NDJSON plumbing.
+ *
+ * <h5 class='topic'>Key types</h5>
+ * <ul>
+ * <li>{@link org.apache.juneau.rest.reactive.ReactiveResponseProcessor} — the response processor
+ * (auto-registered via {@code ServiceLoader}, front-loaded ahead of {@code AsyncResponseProcessor}).
+ * <li>{@link org.apache.juneau.rest.reactive.ReactiveStreamsAdapter} — the {@link java.util.ServiceLoader}
+ * SPI bridge modules implement.
+ * <li>{@link org.apache.juneau.rest.reactive.Adaptation} — the single-value / streaming result of an adapter.
+ * </ul>
+ *
+ * <h5 class='section'>See Also:</h5>
+ * <ul>
+ * <li class='link'><a class="doclink" href="https://juneau.apache.org/docs/topics/RestServerReactive">REST Server — Reactive Streams</a>
+ * <li class='link'><a class="doclink" href="https://www.reactive-streams.org/">Reactive Streams</a>
+ * </ul>
+ *
+ * @since 9.5.0
+ */
+package org.apache.juneau.rest.reactive;
diff --git a/juneau-rest/juneau-rest-server-reactive/src/main/resources/META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor b/juneau-rest/juneau-rest-server-reactive/src/main/resources/META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor
new file mode 100644
index 0000000000..b981efc2f3
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactive/src/main/resources/META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor
@@ -0,0 +1 @@
+org.apache.juneau.rest.reactive.ReactiveResponseProcessor
diff --git a/juneau-rest/juneau-rest-server-reactor/pom.xml b/juneau-rest/juneau-rest-server-reactor/pom.xml
new file mode 100644
index 0000000000..3163152b3a
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactor/pom.xml
@@ -0,0 +1,140 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest</artifactId>
+ <version>9.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>juneau-rest-server-reactor</artifactId>
+ <name>Apache Juneau REST Server Reactive Streams</name>
+ <description>Apache Juneau REST Server - Reactor / RxJava 3 / Reactive-Streams Publisher return-type bridge (opt-in, provided-scope)</description>
+ <packaging>bundle</packaging>
+
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <reactive-streams.version>1.0.4</reactive-streams.version>
+ <reactor.version>3.6.11</reactor.version>
+ <rxjava.version>3.1.10</rxjava.version>
+ </properties>
+
+ <dependencies>
+ <!--
+ Depends on the dependency-free juneau-rest-server-reactive module (which transitively brings
+ juneau-rest-server). That module carries the shared ReactiveResponseProcessor spine + the
+ ReactiveStreamsAdapter SPI + the auto-registration service file; this module contributes ONLY
+ the third-party Reactor / RxJava / Reactive-Streams adapters on top of it.
+ -->
+ <dependency>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest-server-reactive</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!--
+ Reactor / RxJava / Reactive-Streams are ALL declared `provided` so none of them leak as a
+ transitive dependency of this module OR of any module that depends on it. Each adapter is
+ discovered lazily via ServiceLoader and is skipped at runtime if its backing library is
+ absent from the classpath, so a consumer adds `juneau-rest-server-reactor` plus ONLY the
+ reactive library (and version) they actually use.
+
+ Containment guarantee: a `dependency:tree` on juneau-rest-server (the upstream module) must
+ never surface reactor / rxjava / reactive-streams.
+ -->
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
+ <version>${reactive-streams.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.projectreactor</groupId>
+ <artifactId>reactor-core</artifactId>
+ <version>${reactor.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.reactivex.rxjava3</groupId>
+ <artifactId>rxjava</artifactId>
+ <version>${rxjava.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-source-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <phase>verify</phase>
+ <goals>
+ <goal>jar-no-fork</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <extensions>true</extensions>
+ <configuration>
+ <supportIncrementalBuild>true</supportIncrementalBuild>
+ </configuration>
+ <executions>
+ <execution>
+ <id>bundle-manifest</id>
+ <phase>process-classes</phase>
+ <goals>
+ <goal>manifest</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>default-prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>default-report</id>
+ <phase>prepare-package</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/ReactiveStreamsPublisherAdapter.java b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/ReactiveStreamsPublisherAdapter.java
new file mode 100644
index 0000000000..f6ff8f3cfa
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/ReactiveStreamsPublisherAdapter.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive.bridge;
+
+import org.apache.juneau.rest.reactive.*;
+import org.reactivestreams.*;
+
+/**
+ * Catch-all {@link ReactiveStreamsAdapter} for any
+ * <a class="doclink" href="https://www.reactive-streams.org/">Reactive-Streams</a>
+ * {@link org.reactivestreams.Publisher org.reactivestreams.Publisher<T>} that is not already
+ * handled by a more specific adapter.
+ *
+ * <p>
+ * Converts to a JDK {@link java.util.concurrent.Flow.Publisher Flow.Publisher} via
+ * {@link org.reactivestreams.FlowAdapters#toFlowPublisher(Publisher)} and renders it as a stream.
+ * Registered <em>last</em> so library-specific single-value types (Reactor {@code Mono}, which also
+ * implements {@code Publisher}) are matched by their own adapter first.
+ *
+ * @since 9.5.0
+ */
+public class ReactiveStreamsPublisherAdapter implements ReactiveStreamsAdapter {
+
+ @Override /* Overridden from ReactiveStreamsAdapter */
+ public boolean canAdapt(Object value) {
+ return value instanceof Publisher;
+ }
+
+ @Override /* Overridden from ReactiveStreamsAdapter */
+ public Adaptation adapt(Object value) {
+ return Adaptation.stream(FlowAdapters.toFlowPublisher((Publisher<?>) value));
+ }
+}
diff --git a/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/ReactorReactiveAdapter.java b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/ReactorReactiveAdapter.java
new file mode 100644
index 0000000000..2174b7516a
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/ReactorReactiveAdapter.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive.bridge;
+
+import org.apache.juneau.rest.reactive.*;
+import org.reactivestreams.*;
+
+import reactor.core.publisher.*;
+
+/**
+ * {@link ReactiveStreamsAdapter} for <a class="doclink" href="https://projectreactor.io/">Project Reactor</a>
+ * return types.
+ *
+ * <ul>
+ * <li>{@link Mono Mono<T>} → single value (via {@link Mono#toFuture()}). An empty {@code Mono}
+ * completes the response with a {@code null} body.
+ * <li>{@link Flux Flux<T>} → stream (via {@link org.reactivestreams.FlowAdapters#toFlowPublisher(Publisher)}).
+ * </ul>
+ *
+ * <p>
+ * Registered ahead of {@link ReactiveStreamsPublisherAdapter} so a {@code Mono} (which also implements
+ * {@code org.reactivestreams.Publisher}) is treated as a single value rather than a one-element stream.
+ *
+ * @since 9.5.0
+ */
+public class ReactorReactiveAdapter implements ReactiveStreamsAdapter {
+
+ @Override /* Overridden from ReactiveStreamsAdapter */
+ public boolean canAdapt(Object value) {
+ return value instanceof Mono || value instanceof Flux;
+ }
+
+ @Override /* Overridden from ReactiveStreamsAdapter */
+ public Adaptation adapt(Object value) {
+ if (value instanceof Mono<?> m)
+ return Adaptation.single(m.toFuture());
+ var f = (Flux<?>) value;
+ return Adaptation.stream(FlowAdapters.toFlowPublisher(f));
+ }
+}
diff --git a/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/RxJavaReactiveAdapter.java b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/RxJavaReactiveAdapter.java
new file mode 100644
index 0000000000..e7c2b8ed53
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/RxJavaReactiveAdapter.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive.bridge;
+
+import org.apache.juneau.rest.reactive.*;
+import org.reactivestreams.*;
+
+import io.reactivex.rxjava3.core.*;
+
+/**
+ * {@link ReactiveStreamsAdapter} for <a class="doclink" href="https://github.com/ReactiveX/RxJava">RxJava 3</a>
+ * return types.
+ *
+ * <ul>
+ * <li>{@link Single Single<T>} → single value (via {@link Single#toCompletionStage()}).
+ * <li>{@link Maybe Maybe<T>} → single value (via {@link Maybe#toCompletionStage(Object)} with a
+ * {@code null} default, so an empty {@code Maybe} yields a {@code null} body).
+ * <li>{@link Completable} → single {@code null} value (via {@link Completable#toCompletionStage(Object)}).
+ * <li>{@link Flowable Flowable<T>} → stream (via {@link org.reactivestreams.FlowAdapters#toFlowPublisher(Publisher)}).
+ * <li>{@link Observable Observable<T>} → stream (converted to a {@link Flowable} with
+ * {@link BackpressureStrategy#BUFFER} first, since {@code Observable} has no native backpressure).
+ * </ul>
+ *
+ * @since 9.5.0
+ */
+public class RxJavaReactiveAdapter implements ReactiveStreamsAdapter {
+
+ @Override /* Overridden from ReactiveStreamsAdapter */
+ public boolean canAdapt(Object value) {
+ return value instanceof Single
+ || value instanceof Maybe
+ || value instanceof Completable
+ || value instanceof Flowable
+ || value instanceof Observable;
+ }
+
+ @Override /* Overridden from ReactiveStreamsAdapter */
+ public Adaptation adapt(Object value) {
+ if (value instanceof Single<?> s)
+ return Adaptation.single(s.toCompletionStage());
+ if (value instanceof Maybe<?> m)
+ return Adaptation.single(m.toCompletionStage(null));
+ if (value instanceof Completable c)
+ return Adaptation.single(c.toCompletionStage(null));
+ if (value instanceof Flowable<?> f)
+ return Adaptation.stream(FlowAdapters.toFlowPublisher(f));
+ var o = (Observable<?>) value;
+ return Adaptation.stream(FlowAdapters.toFlowPublisher(o.toFlowable(BackpressureStrategy.BUFFER)));
+ }
+}
diff --git a/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/package-info.java b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/package-info.java
new file mode 100644
index 0000000000..6ff5b1d76c
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactor/src/main/java/org/apache/juneau/rest/reactive/bridge/package-info.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Project Reactor / RxJava 3 / Reactive-Streams {@code Publisher} adapters for the
+ * {@code juneau-rest-server} {@link org.apache.juneau.rest.reactive.ReactiveStreamsAdapter} SPI.
+ *
+ * <p>
+ * This module is opt-in. It ships three {@link org.apache.juneau.rest.reactive.ReactiveStreamsAdapter}
+ * implementations — {@link org.apache.juneau.rest.reactive.bridge.ReactorReactiveAdapter},
+ * {@link org.apache.juneau.rest.reactive.bridge.RxJavaReactiveAdapter}, and
+ * {@link org.apache.juneau.rest.reactive.bridge.ReactiveStreamsPublisherAdapter} — registered via a
+ * {@code META-INF/services} file so merely placing this jar (plus the desired reactive library) on the
+ * classpath enables {@code Mono} / {@code Flux} / {@code Single} / {@code Maybe} / {@code Flowable} /
+ * {@code Observable} / {@code Publisher} return types from {@code @RestOp} handlers.
+ *
+ * <h5 class='topic'>Containment</h5>
+ * <p>
+ * The {@code reactor-core}, {@code rxjava}, and {@code reactive-streams} dependencies are declared in
+ * {@code provided} scope on this module's POM. Consumers add only the reactive library and version they
+ * actually use; adapters whose backing library is absent are skipped at {@link java.util.ServiceLoader}
+ * load time. The core {@code juneau-rest-server} jar stays dependency-free.
+ *
+ * <h5 class='section'>See Also:</h5>
+ * <ul>
+ * <li class='jc'>{@link org.apache.juneau.rest.reactive.ReactiveResponseProcessor}
+ * <li class='link'><a class="doclink" href="https://juneau.apache.org/docs/topics/RestServerReactive">REST Server — Reactive Streams</a>
+ * </ul>
+ *
+ * @since 9.5.0
+ */
+package org.apache.juneau.rest.reactive.bridge;
diff --git a/juneau-rest/juneau-rest-server-reactor/src/main/resources/META-INF/services/org.apache.juneau.rest.reactive.ReactiveStreamsAdapter b/juneau-rest/juneau-rest-server-reactor/src/main/resources/META-INF/services/org.apache.juneau.rest.reactive.ReactiveStreamsAdapter
new file mode 100644
index 0000000000..8ff6208546
--- /dev/null
+++ b/juneau-rest/juneau-rest-server-reactor/src/main/resources/META-INF/services/org.apache.juneau.rest.reactive.ReactiveStreamsAdapter
@@ -0,0 +1,3 @@
+org.apache.juneau.rest.reactive.bridge.ReactorReactiveAdapter
+org.apache.juneau.rest.reactive.bridge.RxJavaReactiveAdapter
+org.apache.juneau.rest.reactive.bridge.ReactiveStreamsPublisherAdapter
diff --git a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestContext.java b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestContext.java
index acebf12b8d..ac854c12c5 100644
--- a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestContext.java
+++ b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestContext.java
@@ -1608,6 +1608,59 @@ public class RestContext extends Context {
return u(toList(s));
});
+ /**
+ * JVM-wide cache of {@link ResponseProcessor} provider classes discovered via {@link ServiceLoader}.
+ *
+ * <p>
+ * Populated lazily on first use and shared across all {@link RestContext} instances. Empty on a bare
+ * {@code juneau-rest-server} classpath; non-empty only when an opt-in module (e.g.
+ * {@code juneau-rest-server-reactive}) ships a
+ * {@code META-INF/services/org.apache.juneau.rest.processor.ResponseProcessor} provider file.
+ */
+ private static volatile List<Class<? extends ResponseProcessor>> serviceLoaderResponseProcessors;
+
+ /**
+ * Returns the module-contributed {@link ResponseProcessor} classes discovered via {@link ServiceLoader},
+ * caching the result for the lifetime of the JVM.
+ *
+ * <p>
+ * Provider <em>types</em> are resolved (not instantiated) so each {@link RestContext} can instantiate its
+ * own bean-store-injected copy. Discovery failures are swallowed at {@code FINE} so a malformed provider
+ * never fails resource startup; on a bare classpath the returned list is empty and the default processor
+ * chain is unchanged.
+ *
+ * @return The discovered processor classes (possibly empty, never {@code null}).
+ */
+ private static List<Class<? extends ResponseProcessor>> discoverServiceLoaderResponseProcessors() {
+ var p = serviceLoaderResponseProcessors;
+ if (p == null) {
+ synchronized (RestContext.class) {
+ p = serviceLoaderResponseProcessors;
+ if (p == null) {
+ p = loadServiceLoaderResponseProcessors();
+ serviceLoaderResponseProcessors = p;
+ }
+ }
+ }
+ return p;
+ }
+
+ private static List<Class<? extends ResponseProcessor>> loadServiceLoaderResponseProcessors() {
+ var out = new ArrayList<Class<? extends ResponseProcessor>>();
+ try {
+ for (var it = ServiceLoader.load(ResponseProcessor.class).stream().iterator(); it.hasNext();) {
+ try {
+ out.add(it.next().type());
+ } catch (ServiceConfigurationError | RuntimeException e) {
+ LOG.log(Level.FINE, e, () -> "Skipping ServiceLoader-discovered ResponseProcessor: " + e.getMessage());
+ }
+ }
+ } catch (ServiceConfigurationError | RuntimeException e) {
+ LOG.log(Level.FINE, e, () -> "ServiceLoader for ResponseProcessor failed: " + e.getMessage());
+ }
+ return List.copyOf(out);
+ }
+
/**
* The ordered array of {@link ResponseProcessor} instances for this resource.
*
@@ -1626,6 +1679,12 @@ public class RestContext extends Context {
// at request time when no TracerHook stashed a trace context, so it's zero-cost on the no-tracer path.
if (defaultResponseTraceparent)
b.add(TraceContextResponseProcessor.class);
+ // TODO-119/120 refactor: front-load any module-contributed ResponseProcessors discovered via
+ // ServiceLoader (e.g. the opt-in juneau-rest-server-reactive module's ReactiveResponseProcessor,
+ // which must run ahead of AsyncResponseProcessor). Inert on a bare juneau-rest-server classpath —
+ // when no module ships a META-INF/services/...ResponseProcessor provider file the list is empty and
+ // the chain is identical to the pre-feature default.
+ discoverServiceLoaderResponseProcessors().forEach(b::add);
getRestAnnotationsForProperty(PROPERTY_responseProcessors)
.forEach(ai -> b.add(ai.inner().responseProcessors()));
// @Bean method override REPLACES the entire annotation-derived list.
diff --git a/juneau-rest/pom.xml b/juneau-rest/pom.xml
index ca60da7b71..eb4523c9b4 100644
--- a/juneau-rest/pom.xml
+++ b/juneau-rest/pom.xml
@@ -42,6 +42,8 @@
<module>juneau-rest-server-oidc-rp</module>
<module>juneau-rest-server-micrometer</module>
<module>juneau-rest-server-otel</module>
+ <module>juneau-rest-server-reactive</module>
+ <module>juneau-rest-server-reactor</module>
<module>juneau-rest-server-view-jsp</module>
<module>juneau-rest-server-view-thymeleaf</module>
<module>juneau-rest-server-view-mustache</module>
diff --git a/juneau-utest/pom.xml b/juneau-utest/pom.xml
index 1a30065ec0..615921661f 100644
--- a/juneau-utest/pom.xml
+++ b/juneau-utest/pom.xml
@@ -541,6 +541,55 @@
<version>1.62.0</version>
<scope>test</scope>
</dependency>
+ <!--
+ Reactive-Streams return-type modules (both opt-in; the reactive feature is no longer in
+ juneau-rest-server core / DefaultConfig — see FINISHED-119/120).
+
+ - juneau-rest-server-reactive: dependency-free JDK Flow.Publisher spine + ReactiveStreamsAdapter
+ SPI + the ServiceLoader auto-registration file for ReactiveResponseProcessor. Its mere presence
+ on this test classpath is what activates the reactive processor for ReactiveResponseProcessor_Test
+ (the JDK-native path needs no third-party library). Pulled transitively by reactor, but declared
+ explicitly here for clarity and so the JDK-Flow tests don't silently depend on reactor.
+ - juneau-rest-server-reactor: the Reactor / RxJava / Reactive-Streams adapter bridge. The module
+ declares reactor-core / rxjava / reactive-streams in `provided` scope (so they never leak into
+ consumers); juneau-utest exercises all three adapters end-to-end, so we add the runtime libs here
+ in `test` scope explicitly.
+
+ NOTE: because juneau-rest-server-reactive is on this classpath, EVERY resource in juneau-utest gets
+ the ReactiveResponseProcessor auto-discovered. The "bare juneau-rest-server has zero reactive
+ behavior" assertion therefore cannot live here — it is exercised in juneau-rest-mock (whose test
+ classpath has no reactive module). See ReactiveOptIn_BareServer_Test.
+ -->
+ <dependency>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest-server-reactive</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.juneau</groupId>
+ <artifactId>juneau-rest-server-reactor</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
+ <version>1.0.4</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.projectreactor</groupId>
+ <artifactId>reactor-core</artifactId>
+ <version>3.6.11</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.reactivex.rxjava3</groupId>
+ <artifactId>rxjava</artifactId>
+ <version>3.1.10</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.openjdk.jmh</groupId>
<artifactId>jmh-core</artifactId>
diff --git a/juneau-utest/src/test/java/org/apache/juneau/rest/reactive/ReactiveResponseProcessor_Test.java b/juneau-utest/src/test/java/org/apache/juneau/rest/reactive/ReactiveResponseProcessor_Test.java
new file mode 100644
index 0000000000..ba42ab1cd8
--- /dev/null
+++ b/juneau-utest/src/test/java/org/apache/juneau/rest/reactive/ReactiveResponseProcessor_Test.java
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+import org.apache.juneau.TestBase;
+import org.apache.juneau.json.*;
+import org.apache.juneau.rest.*;
+import org.apache.juneau.rest.annotation.*;
+import org.apache.juneau.rest.mock.classic.*;
+import org.apache.juneau.sse.*;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests {@link ReactiveResponseProcessor} using only the JDK-native
+ * {@link java.util.concurrent.Flow.Publisher Flow.Publisher} return type (no external reactive
+ * library). This is the MAYBE-120 spine: {@code Flow.Publisher<SseEvent>} SSE streaming plus the
+ * shared buffer / NDJSON shapes that all reactive types funnel through.
+ *
+ * <p>
+ * Under {@code MockRestClient} the underlying {@code MockServletRequest} reports
+ * {@code isAsyncSupported() == false}, so streaming exercises its synchronous-fallback path (subscribe
+ * and block until the publisher terminates, writing frames as they arrive) and buffering exercises the
+ * {@code AsyncResponseProcessor} synchronous fallback.
+ */
+class ReactiveResponseProcessor_Test extends TestBase {
+
+ public static final class Pojo {
+ public final String name;
+ public final int value;
+ public Pojo(String name, int value) { this.name = name; this.value = value; }
+ }
+
+ /**
+ * Trampolined synchronous {@link java.util.concurrent.Flow.Publisher Flow.Publisher} that replays a
+ * fixed list on subscribe, honoring {@code request(n)} without recursing on re-request (so the
+ * one-at-a-time streaming subscriber is exercised deterministically on a single thread).
+ */
+ static final class ListPublisher<T> implements Flow.Publisher<T> {
+ private final List<T> items;
+ ListPublisher(List<T> items) { this.items = items; }
+
+ @SafeVarargs
+ static <T> ListPublisher<T> of(T... items) { return new ListPublisher<>(List.of(items)); }
+
+ @Override public void subscribe(Flow.Subscriber<? super T> sub) {
+ sub.onSubscribe(new Flow.Subscription() {
+ private int idx;
+ private final AtomicLong demand = new AtomicLong();
+ private boolean draining;
+ private volatile boolean cancelled;
+
+ @Override public void request(long n) {
+ if (cancelled)
+ return;
+ demand.addAndGet(n);
+ if (draining)
+ return;
+ draining = true;
+ try {
+ while (! cancelled && demand.get() > 0 && idx < items.size()) {
+ demand.decrementAndGet();
+ sub.onNext(items.get(idx++));
+ }
+ if (! cancelled && idx >= items.size())
+ sub.onComplete();
+ } finally {
+ draining = false;
+ }
+ }
+
+ @Override public void cancel() { cancelled = true; }
+ });
+ }
+ }
+
+ /** Publisher that emits one element then errors. */
+ static final class ErrorPublisher<T> implements Flow.Publisher<T> {
+ private final T first;
+ private final boolean emitFirst;
+ ErrorPublisher(T first, boolean emitFirst) { this.first = first; this.emitFirst = emitFirst; }
+
+ @Override public void subscribe(Flow.Subscriber<? super T> sub) {
+ sub.onSubscribe(new Flow.Subscription() {
+ private boolean done;
+ @Override public void request(long n) {
+ if (done)
+ return;
+ done = true;
+ if (emitFirst)
+ sub.onNext(first);
+ sub.onError(new RuntimeException("boom"));
+ }
+ @Override public void cancel() { done = true; }
+ });
+ }
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // A: Buffer shape — default media type collects the stream into a JSON array.
+ // -----------------------------------------------------------------------------------------------------------------
+
+ @Rest(serializers = JsonSerializer.class)
+ public static class A {
+ @RestGet("/flux")
+ public Flow.Publisher<Pojo> flux() {
+ return ListPublisher.of(new Pojo("foo", 1), new Pojo("bar", 2));
+ }
+
+ @RestGet("/empty")
+ public Flow.Publisher<Pojo> empty() {
+ return ListPublisher.of();
+ }
+
+ @RestGet("/sse")
+ public Flow.Publisher<SseEvent> sse(RestResponse res) {
+ res.setContentType("text/event-stream");
+ return ListPublisher.of(new SseEvent("tick", "one"), new SseEvent("tick", "two"));
+ }
+
+ @RestGet("/sseNonEvent")
+ public Flow.Publisher<Pojo> sseNonEvent(RestResponse res) {
+ res.setContentType("text/event-stream");
+ return ListPublisher.of(new Pojo("foo", 1));
+ }
+
+ @RestGet("/ndjson")
+ public Flow.Publisher<Pojo> ndjson(RestResponse res) {
+ res.setContentType("application/x-ndjson");
+ return ListPublisher.of(new Pojo("foo", 1), new Pojo("bar", 2));
+ }
+
+ @RestGet("/errStream")
+ public Flow.Publisher<SseEvent> errStream(RestResponse res) {
+ res.setContentType("text/event-stream");
+ return new ErrorPublisher<>(new SseEvent("tick", "one"), true);
+ }
+ }
+
+ private static final MockRestClient CA = MockRestClient.buildLax(A.class);
+
+ @Test void a01_flux_buffersToJsonArray() throws Exception {
+ CA.get("/flux").accept("application/json").run()
+ .assertStatus(200)
+ .assertContent().isContains("\"name\":\"foo\"", "\"value\":1", "\"name\":\"bar\"", "\"value\":2");
+ }
+
+ @Test void a02_emptyFlux_buffersToEmptyArray() throws Exception {
+ CA.get("/empty").accept("application/json").run().assertStatus(200).assertContent().is("[]");
+ }
+
+ @Test void a03_flowPublisherOfSseEvent_streamsSse() throws Exception {
+ var r = CA.get("/sse").run();
+ r.assertStatus(200);
+ r.assertHeader("Content-Type").isContains("text/event-stream");
+ var c = r.getContent().asString();
+ assertTrue(c.contains("event: tick"), c);
+ assertTrue(c.contains("data: one"), c);
+ assertTrue(c.contains("data: two"), c);
+ }
+
+ @Test void a04_sseWithNonEventElement_jsonEncodesData() throws Exception {
+ var c = CA.get("/sseNonEvent").run().assertStatus(200).getContent().asString();
+ assertTrue(c.contains("data:"), c);
+ assertTrue(c.contains("\"name\":\"foo\""), c);
+ }
+
+ @Test void a05_ndjson_writesOneJsonObjectPerLine() throws Exception {
+ var c = CA.get("/ndjson").run().assertStatus(200).getContent().asString();
+ var lines = c.strip().split("\n");
+ assertEquals(2, lines.length, c);
+ assertTrue(lines[0].contains("\"name\":\"foo\""), c);
+ assertTrue(lines[1].contains("\"name\":\"bar\""), c);
+ }
+
+ @Test void a06_streamErrorAfterFirstFrame_stillDeliversFirstFrame() throws Exception {
+ // First frame is committed before the error; SSE has no way to retract it, so the first
+ // event is delivered and the stream simply ends.
+ var c = CA.get("/errStream").run().getContent().asString();
+ assertTrue(c.contains("data: one"), c);
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // B: Accept-header-driven SSE opt-in (serializer registered so negotiation passes).
+ // -----------------------------------------------------------------------------------------------------------------
+
+ @Rest(serializers = SseSerializer.class)
+ public static class B {
+ @RestGet("/events")
+ public Flow.Publisher<SseEvent> events() {
+ return ListPublisher.of(new SseEvent("msg", "hello"));
+ }
+ }
+
+ private static final MockRestClient CB = MockRestClient.buildLax(B.class);
+
+ @Test void b01_acceptHeaderSelectsSse() throws Exception {
+ var r = CB.get("/events").header("Accept", "text/event-stream").run();
+ r.assertStatus(200);
+ r.assertHeader("Content-Type").isContains("text/event-stream");
+ assertTrue(r.getContent().asString().contains("data: hello"));
+ }
+
+ // -----------------------------------------------------------------------------------------------------------------
+ // C: Non-reactive return values are untouched (processor is a transparent no-op).
+ // -----------------------------------------------------------------------------------------------------------------
+
+ @Rest(serializers = JsonSerializer.class)
+ public static class C {
+ @RestGet("/plain")
+ public Pojo plain() {
+ return new Pojo("plain", 99);
+ }
+
+ @RestGet("/future")
+ public CompletableFuture<Pojo> future() {
+ return CompletableFuture.completedFuture(new Pojo("fut", 7));
+ }
+ }
+
+ private static final MockRestClient CC = MockRestClient.buildLax(C.class);
+
+ @Test void c01_plainPojo_unaffected() throws Exception {
+ CC.get("/plain").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"plain\"", "\"value\":99");
+ }
+
+ @Test void c02_completableFuture_stillHandledByAsyncProcessor() throws Exception {
+ CC.get("/future").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"fut\"", "\"value\":7");
+ }
+}
diff --git a/juneau-utest/src/test/java/org/apache/juneau/rest/reactive/bridge/ReactiveBridge_Test.java b/juneau-utest/src/test/java/org/apache/juneau/rest/reactive/bridge/ReactiveBridge_Test.java
new file mode 100644
index 0000000000..704ffab9be
--- /dev/null
+++ b/juneau-utest/src/test/java/org/apache/juneau/rest/reactive/bridge/ReactiveBridge_Test.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.juneau.rest.reactive.bridge;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.List;
+import java.util.concurrent.atomic.*;
+
+import org.apache.juneau.TestBase;
+import org.apache.juneau.json.*;
+import org.apache.juneau.rest.*;
+import org.apache.juneau.rest.annotation.*;
+import org.apache.juneau.rest.mock.classic.*;
+import org.apache.juneau.sse.*;
+import org.junit.jupiter.api.Test;
+
+import io.reactivex.rxjava3.core.*;
+import reactor.core.publisher.*;
+
+/**
+ * End-to-end tests for the {@code juneau-rest-server-reactor} bridge adapters
+ * ({@link ReactorReactiveAdapter}, {@link RxJavaReactiveAdapter},
+ * {@link ReactiveStreamsPublisherAdapter}) wired through the core
+ * {@link org.apache.juneau.rest.reactive.ReactiveResponseProcessor} via ServiceLoader.
+ */
+class ReactiveBridge_Test extends TestBase {
+
+ public static final class Pojo {
+ public final String name;
+ public final int value;
+ public Pojo(String name, int value) { this.name = name; this.value = value; }
+ }
+
+ /** Minimal trampolined Reactive-Streams publisher, used to exercise the generic Publisher adapter. */
+ static final class RsListPublisher implements org.reactivestreams.Publisher<Pojo> {
+ private final List<Pojo> items;
+ RsListPublisher(List<Pojo> items) { this.items = items; }
+
+ @Override public void subscribe(org.reactivestreams.Subscriber<? super Pojo> sub) {
+ sub.onSubscribe(new org.reactivestreams.Subscription() {
+ private int idx;
+ private final AtomicLong demand = new AtomicLong();
+ private boolean draining;
+ private volatile boolean cancelled;
+
+ @Override public void request(long n) {
+ if (cancelled)
+ return;
+ demand.addAndGet(n);
+ if (draining)
+ return;
+ draining = true;
+ try {
+ while (! cancelled && demand.get() > 0 && idx < items.size()) {
+ demand.decrementAndGet();
+ sub.onNext(items.get(idx++));
+ }
+ if (! cancelled && idx >= items.size())
+ sub.onComplete();
+ } finally {
+ draining = false;
+ }
+ }
+
+ @Override public void cancel() { cancelled = true; }
+ });
+ }
+ }
+
+ @Rest(serializers = JsonSerializer.class)
+ public static class A {
+
+ // ---- Reactor ----
+ @RestGet("/mono")
+ public Mono<Pojo> mono() { return Mono.just(new Pojo("mono", 1)); }
+
+ @RestGet("/monoEmpty")
+ public Mono<Pojo> monoEmpty() { return Mono.empty(); }
+
+ @RestGet("/flux")
+ public Flux<Pojo> flux() { return Flux.just(new Pojo("a", 1), new Pojo("b", 2)); }
+
+ @RestGet("/fluxSse")
+ public Flux<SseEvent> fluxSse(RestResponse res) {
+ res.setContentType("text/event-stream");
+ return Flux.just(new SseEvent("tick", "one"), new SseEvent("tick", "two"));
+ }
+
+ // ---- RxJava 3 ----
+ @RestGet("/single")
+ public Single<Pojo> single() { return Single.just(new Pojo("single", 3)); }
+
+ @RestGet("/maybe")
+ public Maybe<Pojo> maybe() { return Maybe.just(new Pojo("maybe", 4)); }
+
+ @RestGet("/maybeEmpty")
+ public Maybe<Pojo> maybeEmpty() { return Maybe.empty(); }
+
+ @RestGet("/completable")
+ public Completable completable() { return Completable.complete(); }
+
+ @RestGet("/flowable")
+ public Flowable<Pojo> flowable() { return Flowable.just(new Pojo("x", 5), new Pojo("y", 6)); }
+
+ @RestGet("/observable")
+ public Observable<Pojo> observable() { return Observable.just(new Pojo("o", 7), new Pojo("p", 8)); }
+
+ // ---- Generic Reactive-Streams Publisher ----
+ @RestGet("/publisher")
+ public org.reactivestreams.Publisher<Pojo> publisher() {
+ return new RsListPublisher(List.of(new Pojo("pub", 9)));
+ }
+ }
+
+ private static final MockRestClient C = MockRestClient.buildLax(A.class);
+
+ // ---- Reactor ----
+
+ @Test void a01_mono_single() throws Exception {
+ C.get("/mono").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"mono\"", "\"value\":1");
+ }
+
+ @Test void a02_monoEmpty_nullBody() throws Exception {
+ C.get("/monoEmpty").accept("application/json").run().assertStatus(200);
+ }
+
+ @Test void a03_flux_buffersToArray() throws Exception {
+ C.get("/flux").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"a\"", "\"name\":\"b\"");
+ }
+
+ @Test void a04_flux_streamsSse() throws Exception {
+ var c = C.get("/fluxSse").run().assertStatus(200).getContent().asString();
+ assertTrue(c.contains("event: tick"), c);
+ assertTrue(c.contains("data: one"), c);
+ assertTrue(c.contains("data: two"), c);
+ }
+
+ // ---- RxJava 3 ----
+
+ @Test void b01_single() throws Exception {
+ C.get("/single").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"single\"", "\"value\":3");
+ }
+
+ @Test void b02_maybe() throws Exception {
+ C.get("/maybe").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"maybe\"", "\"value\":4");
+ }
+
+ @Test void b03_maybeEmpty_nullBody() throws Exception {
+ C.get("/maybeEmpty").accept("application/json").run().assertStatus(200);
+ }
+
+ @Test void b04_completable_nullBody() throws Exception {
+ C.get("/completable").accept("application/json").run().assertStatus(200);
+ }
+
+ @Test void b05_flowable_buffersToArray() throws Exception {
+ C.get("/flowable").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"x\"", "\"name\":\"y\"");
+ }
+
+ @Test void b06_observable_buffersToArray() throws Exception {
+ C.get("/observable").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"o\"", "\"name\":\"p\"");
+ }
+
+ // ---- Generic Publisher ----
+
+ @Test void c01_reactiveStreamsPublisher_buffersToArray() throws Exception {
+ C.get("/publisher").accept("application/json").run().assertStatus(200)
+ .assertContent().isContains("\"name\":\"pub\"", "\"value\":9");
+ }
+}
diff --git a/juneau-utest/test-run-history.tsv b/juneau-utest/test-run-history.tsv
index e632c8c64f..7ce105bf82 100644
--- a/juneau-utest/test-run-history.tsv
+++ b/juneau-utest/test-run-history.tsv
@@ -54,3 +54,4 @@ timestamp git_sha branch tests_run failures errors skipped surefire_sec wall_sec
2026-05-29T10:45:54Z 3dc947b24d41 master 125897 0 0 21 151
2026-05-29T13:48:36Z c59fc14fac5e master 126037 0 0 21 144
2026-05-29T14:57:25Z 830e0ca051c6 master 126043 0 0 21 149
+2026-05-29T16:30:02Z 29bd5a598e5a master 126063 0 0 21 148