/*
 * 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.server.reactive;

import static jakarta.servlet.http.HttpServletResponse.*;
import static org.apache.juneau.commons.utils.Shorts.*;

import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.*;
import java.util.logging.*;

import org.apache.juneau.commons.http.MediaRanges;
import org.apache.juneau.http.response.*;
import org.apache.juneau.marshall.marshaller.*;
import org.apache.juneau.marshall.serializer.*;
import org.apache.juneau.marshall.sse.*;
import org.apache.juneau.rest.server.*;
import org.apache.juneau.rest.server.logging.*;
import org.apache.juneau.rest.server.processor.*;
import org.apache.juneau.rest.server.util.*;

import jakarta.servlet.*;
import jakarta.servlet.http.*;

/**
 * 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/ResponseProcessor} provider file is discovered
 * by {@code RestContext} via {@link ServiceLoader} and this processor is front-loaded ahead of
 * {@code AsyncResponseProcessor} in the chain &mdash; 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&lt;T&gt;} (no external dependency
 * required) and, through registered {@link ReactiveStreamsAdapter} providers, any third-party reactive
 * type the opt-in {@code juneau-rest-server-reactive-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}) &mdash; 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}, {@code application/jsonlines},
 *       {@code application/json5l}, {@code application/json5lines}, {@code application/json-seq}) &mdash; each
 *       element is JSON-encoded on its own line. Unless the handler set a Content-Type, the response echoes the
 *       highest-ranked NDJSON-family {@code Accept} range in canonical form ({@code jsonlines} becomes
 *       {@code jsonl}, {@code json5lines} becomes {@code json5l}, {@code json-seq} becomes {@code x-ndjson});
 *       otherwise it is {@code application/x-ndjson}. See {@link #NDJSON_ALIASES}.
 *    <li><b>Buffer</b> (default, any other media type) &mdash; all elements are collected into a
 *       {@link 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 &mdash; 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)} 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 {@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 10.0.0
 */
@SuppressWarnings({
   "java:S1192", // Media-type literals are protocol tokens; the alias table reads best with them inline
   "java:S2095", // Writer is response-owned; we never close it.
   "resource" // Closeables here are framework-managed and not owned/closed by this class; not a real leak.
})
public class ReactiveResponseProcessor implements ResponseProcessor {

   private static final Logger LOG = Logger.getLogger(ReactiveResponseProcessor.class.getName());

   private static final AtomicReference<List<ReactiveStreamsAdapter>> ADAPTERS = new AtomicReference<>();

   private enum Shape { BUFFER, SSE, NDJSON }

   /**
    * NDJSON-family media types accepted from {@code Accept}, mapped to the Content-Type echoed on an NDJSON-shape
    * response. {@code application/json-seq} maps to {@code application/x-ndjson} because this writer does not emit
    * RFC 7464 record separators.
    */
   static final Map<String,String> NDJSON_ALIASES = Map.of(
      "application/jsonl", "application/jsonl",
      "application/jsonlines", "application/jsonl",
      "application/json5l", "application/json5l",
      "application/json5lines", "application/json5l",
      "application/x-ndjson", "application/x-ndjson",
      "application/json-seq", "application/x-ndjson"
   );

   /**
    * Returns the canonical NDJSON-family type of the highest-ranked {@code Accept} range in that family.
    *
    * <p>
    * Ranges are walked in {@link MediaRanges} order (q descending, then its own tie-break). Ranges with
    * {@code q=0} are skipped; parameters other than {@code q} are ignored; matching is case-insensitive.
    *
    * @param accept The raw {@code Accept} header, possibly <jk>null</jk>.
    * @return The canonical type, or <jk>null</jk> when no NDJSON-family range is acceptable.
    */
   static String negotiatedNdjsonType(String accept) {
      for (var r : MediaRanges.of(accept).toList()) {
         if (r.getQValue() <= 0f)
            continue;
         var echoed = NDJSON_ALIASES.get((r.getType() + "/" + r.getSubType()).toLowerCase(Locale.ROOT));
         if (echoed != null)
            return echoed;
      }
      return null;
   }

   @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<?> content2)
         return dispatch(opSession, content2);

      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:S1141", // Nested try/catch cleanly separates startAsync recovery from the sync fallback.
      "java:S3776" // Streaming setup is branchy by nature (content-type prep + async-vs-sync fallback).
   })
   private int handleStream(RestOpSession opSession, Flow.Publisher<?> pub, Shape shape) throws IOException {
      var res = opSession.getResponse();
      prepareStreamingHeaders(res, shape, opSession.getRequest().getHttpServletRequest().getHeader("Accept"));

      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) {
         // Publish async ownership + the resolved debug snapshot on the request thread BEFORE subscribing, so an
         // inline-terminating publisher never observes a null snapshot and finish() reliably skips synchronous emit.
         AsyncResponseProcessor.markAsyncOwnedAndSnapshot(opSession);
         var ac = asyncCtx;
         ac.setTimeout(0);  // No artificial timeout — streams are paced by the producer / client disconnect.

         // Single exactly-once completion gate shared by the subscriber terminal, the container AsyncListener
         // fallback (error/timeout/complete), and a subscribe() that throws after handoff — so a client abort or
         // container error can never silently skip the completion-path emit, and it can never fire twice.
         var completed = new AtomicBoolean(false);
         Consumer<Throwable> completion = t -> {
            if (completed.compareAndSet(false, true))
               completeAsync(ac, opSession, t);
         };
         ac.addListener(new ReactiveCompletionListener(completion));
         try {
            pub.subscribe(new StreamingSubscriber(res, writer, encoder, mdc, completion));
         } catch (RuntimeException e) {
            // subscribe() threw after the async handoff — route it through the same completion hook rather than
            // leaking async ownership with no emit and no AsyncContext.complete().
            completion.accept(e);
         }
         return FINISHED;
      }

      // Synchronous fallback (MockServletRequest et al.): block until the publisher terminates.
      var done = new CompletableFuture<Void>();
      var subscriber = new StreamingSubscriber(res, writer, encoder, mdc,
         t -> { if (t == null) done.complete(null); else done.completeExceptionally(t); });
      pub.subscribe(subscriber);
      if (! awaitSync(done, syncTimeoutMillis(opSession)))
         // Timed out or interrupted before the publisher terminated — cancel the still-live subscription so it
         // cannot emit late writes against a response we are about to finish.
         subscriber.cancel();
      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, String accept) {
      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) {
         var echoed = negotiatedNdjsonType(accept);
         res.setContentType(echoed == null ? "application/x-ndjson" : echoed);
      }
      res.setHeader("Cache-Control", "no-cache");
      res.setHeader("Content-Encoding", "identity");
   }

   /**
    * Blocks until the synchronous stream completes, times out, or is interrupted.
    *
    * @return <jk>true</jk> if the publisher terminated (normally or with an error) before the timeout; <jk>false</jk>
    *    if the wait timed out or was interrupted (in which case the caller must cancel the still-live subscription).
    */
   @SuppressWarnings({
      "java:S2142" // Interrupt flag restored before returning.
   })
   private static boolean awaitSync(CompletableFuture<Void> done, long timeoutMs) {
      try {
         done.get(timeoutMs, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         return false;
      } catch (TimeoutException e) {
         LOG.log(Level.FINE, e, () -> "Synchronous reactive stream timed out: " + e.getMessage());
         return false;
      } catch (ExecutionException e) {
         // Publisher terminated with an error — the subscription is already finished, no orphan to cancel.
         LOG.log(Level.FINE, e, () -> "Synchronous reactive stream did not complete cleanly: " + e.getMessage());
      }
      return true;
   }

   private static void completeAsync(AsyncContext asyncCtx, RestOpSession opSession, Throwable error) {
      var res = opSession.getResponse();
      var session = opSession.getRestSession();
      try {
         if (error != null) {
            LOG.log(Level.FINE, error, () -> "Reactive stream terminated with error: " + error.getMessage());
            session.exception(error);
         }
         if (! res.getHttpServletResponse().isCommitted())
            res.flushBuffer();
      } catch (IOException e) {
         LOG.log(Level.FINEST, e, () -> "Flush during async stream completion failed: " + e.getMessage());
      } finally {
         // Same completion-path ordering as ordinary async: render from the final stream state, then release the
         // request, then complete the async context. Nested try/finally isolates each step so a debug-emit failure
         // cannot suppress request close, and a request-close failure cannot suppress async-context completion.
         try {
            RestDebugPipeline.emitOnCompletion(session);
         } finally {
            try {
               opSession.closeRequest();
            } catch (RuntimeException e) { // HTT: defensive; the guarded call does not throw today
               LOG.log(Level.FINEST, e, () -> "Async request close failed: " + e.getMessage());
            } finally {
               completeQuietly(asyncCtx);
            }
         }
      }
   }

   private static void completeQuietly(AsyncContext asyncCtx) {
      try {
         asyncCtx.complete();
      } catch (IllegalStateException e) {
         LOG.log(Level.FINEST, e, () -> "AsyncContext.complete() raced with the container: " + e.getMessage());
      }
   }

   /**
    * Container-side fallback that routes servlet async terminal events (error, timeout, complete) through the shared
    * completion hook.
    *
    * <p>
    * The subscriber's own terminal callback is the normal completion trigger; this listener guarantees that a container
    * error or timeout that bypasses the subscriber still reaches the exactly-once completion gate, so completion-path
    * emission and request close are never silently skipped. All events funnel through the same gated {@link Consumer};
    * {@code onComplete} (fired by our own {@code AsyncContext.complete()}) is therefore a gated no-op.
    */
   private static final class ReactiveCompletionListener implements AsyncListener {
      private final Consumer<Throwable> completion;

      ReactiveCompletionListener(Consumer<Throwable> completion) {
         this.completion = completion;
      }

      @Override public void onComplete(AsyncEvent event) { completion.accept(null); }
      @Override public void onError(AsyncEvent event) { completion.accept(event.getThrowable()); }
      @Override public void onTimeout(AsyncEvent event) { completion.accept(new TimeoutException("Reactive stream timed out.")); }
      @Override public void onStartAsync(AsyncEvent event) { /* no-op */ }
   }

   // -----------------------------------------------------------------------------------------------------------------
   // Frame encoders.
   // -----------------------------------------------------------------------------------------------------------------

   private static void writeSseFrame(FinishablePrintWriter w, Object element) throws IOException, SerializeException {
      if (element == null)
         return;
      if (element instanceof SseEvent element2) {
         Sse.DEFAULT.write(element2, w);
         return;
      }
      var data = element instanceof CharSequence element2 ? element2.toString() : Json.of(element);
      Sse.DEFAULT.write(new SseEvent(null, data), w);
   }

   private static void writeNdjsonFrame(FinishablePrintWriter w, Object element) throws SerializeException {
      if (element == null)
         return;
      var json = element instanceof CharSequence element2 ? element2.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("json5l") || 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.get();
      if (a == null) {
         synchronized (ReactiveResponseProcessor.class) {
            a = ADAPTERS.get();
            if (a == null) { // HTT: inner DCL null-check missed branch requires concurrent first-load race
               a = loadAdapters();
               ADAPTERS.set(a);
            }
         }
      }
      return a;
   }

   private static List<ReactiveStreamsAdapter> loadAdapters() {
      var out = new ArrayList<ReactiveStreamsAdapter>();
      try {
         var it = ServiceLoader.load(ReactiveStreamsAdapter.class).iterator();
         while (it.hasNext())
            addNextAdapter(it, out);
      } catch (ServiceConfigurationError | RuntimeException e) { // HTT: defensive; a malformed registration file fails inside it.next(), which addNextAdapter handles, so this only triggers if hasNext() itself fails (not reproducible without breaking ServiceLoader for the other tests)
         LOG.log(Level.FINE, e, () -> "ServiceLoader for ReactiveStreamsAdapter failed: " + e.getMessage());
      }
      return List.copyOf(out);
   }

   private static void addNextAdapter(Iterator<ReactiveStreamsAdapter> it, List<ReactiveStreamsAdapter> out) {
      try {
         out.add(it.next());
      } catch (ServiceConfigurationError | RuntimeException e) {
         LOG.log(Level.FINE, e, () -> "Skipping reactive adapter (backing library likely absent): " + e.getMessage()); // HTT: adapters are loaded once per JVM, before any test can attach a log handler, so this message supplier only runs when FINE logging is on at first use; a seam (resettable adapter cache) would make this testable
      }
   }

   // -----------------------------------------------------------------------------------------------------------------
   // 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 = l();

      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;

      /** Published once by {@code onSubscribe} (Reactive Streams: at most one call); read by {@code onNext}/{@code cancel}. */
      private final AtomicReference<Flow.Subscription> subscription = new AtomicReference<>();

      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.set(s);
         s.request(1);
      }

      @Override public void onNext(Object item) {
         try {
            withMdc(() -> writeFrame(item));
            subscription.get().request(1);
         } catch (RuntimeException e) {
            subscription.get().cancel();
            terminate(unwrap(e));
         }
      }

      @Override public void onError(Throwable t) { terminate(t); }
      @Override public void onComplete() { terminate(null); }

      /**
       * Cancels the underlying subscription and marks this subscriber terminated so no late frames are written.
       *
       * <p>
       * Invoked by the synchronous fallback path when the wait times out or is interrupted before the publisher
       * terminates, preventing an orphaned subscription from emitting writes against a finished response.
       */
      void cancel() {
         if (! terminated.compareAndSet(false, true)) // HTT: true branch needs the publisher to terminate between the wait timing out and cancel(), a cross-thread race; a seam would make this testable
            return;
         var s = subscription.get();
         if (s != null)
            s.cancel();
      }

      private void writeFrame(Object item) {
         try {
            encoder.write(writer, item);
            writer.flush();
            // PrintWriter (which writer extends) never lets an IOException from the underlying stream escape —
            // it swallows it and only records it via checkError(). Without this check, a broken/closed client
            // connection is silently absorbed here and the pump keeps "succeeding" forever instead of detecting
            // the write failure and cleaning up (the dead-client cleanup path, see SEP-2243).
            if (writer.checkError())
               throw new IOException("Write failed: the client connection is no longer writable.");
            res.flushBuffer();
            wrote = true;
         } catch (IOException | SerializeException e) {
            throw isex(e);
         }
      }

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