Package org.apache.juneau.rest.server.reactive

Class ReactiveResponseProcessor

java.lang.Object
org.apache.juneau.rest.server.reactive.ReactiveResponseProcessor
All Implemented Interfaces:
ResponseProcessor
public class ReactiveResponseProcessor extends Object implements ResponseProcessor
Response processor that bridges Reactive Streams return values from @RestOp handlers to Juneau's response pipeline.
Activation (opt-in)

This processor ships in the opt-in juneau-rest-server-reactive module and is not wired into DefaultConfig. A bare juneau-rest-server has zero reactive behavior. When the juneau-rest-server-reactive jar is on the classpath, its META-INF/services/ResponseProcessor provider file is discovered by RestContext via ServiceLoader and this processor is front-loaded ahead of AsyncResponseProcessor in the chain — no @Rest(responseProcessors=...) entry required.

What it handles

This is the single, shared spine for all reactive return-type support. It natively understands the JDK type Flow.Publisher<T> (no external dependency required) and, through registered ReactiveStreamsAdapter providers, any third-party reactive type the opt-in juneau-rest-server-reactive-reactor module adapts to it (Project Reactor Mono / Flux, RxJava 3 Single / Maybe / Flowable / Observable, and the Reactive-Streams org.reactivestreams.Publisher).

Single-value reactive types (e.g. Mono, Single) are adapted to a CompletionStage and collapse onto the existing AsyncResponseProcessor async path, inheriting its timeout, completion-executor (@Rest(asyncCompletionExecutor)), and MDC bridging behavior for free.

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, then the request Accept header):

  • SSE (text/event-stream) — each element is 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/jsonlines, application/json5l, application/json5lines, application/json-seq) — each element is JSON-encoded on its own line. Unless the handler set a Content-Type, the response echoes the highest-ranked NDJSON-family Accept range in canonical form (jsonlines becomes jsonl, json5lines becomes json5l, json-seq becomes x-ndjson); otherwise it is application/x-ndjson. See NDJSON_ALIASES.
  • Buffer (default, any other media type) — all elements are 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.
Backpressure

Streaming subscribers request one element at a time (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 Long.MAX_VALUE since the collection is bounded by the publisher's own completion.

Threading, executors, and MDC

Buffer-shape responses route through AsyncResponseProcessor and therefore honor @Rest(asyncCompletionExecutor) and the SLF4J MDC bridge. 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 scheduler model. When MDC propagation is enabled, the request-thread MDC snapshot is reinstalled around each onNext / terminal callback via MdcAsyncListener so log statements emitted while writing a frame see the request's diagnostic context.

Synchronous fallback

In environments where ServletRequest.startAsync() is unsupported (notably Juneau's 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.

Since:
10.0.0
See Also:

  • Constructor Details

  • Method Details

    • process

      public int process(RestOpSession opSession) throws IOException, BasicHttpException
      Description copied from interface: ResponseProcessor
      Process this response if possible.
      Specified by:
      process in interface ResponseProcessor
      Parameters:
      opSession - The HTTP call.
      Returns:
      One of the following codes:
      • 0 - The processor could not handle the request.
      • 1 - The processor was able to fully handle the request.
      • 2 - The processor was able to partially handle the request by replacing the output. The response processors should start over.
      Throws:
      IOException - If low-level exception occurred on output stream.
      Results in a HttpServletResponse.SC_INTERNAL_SERVER_ERROR error.
      BasicHttpException - If some other exception occurred.
      Can be used to provide an appropriate HTTP response code and message.