Fixes #12122 - NPE in HttpReceiver.responseContentAvailable().

Now also the HttpReceiver.responseContentAvailable() is serialized, so that the access to `this.contentSource` is serialized with failure, and protected by a call to `exchange.isResponseCompleteOrTerminated()`.

Before, it was possible that a thread failed the response, nulling out `this.contentSource`, while another thread was just about to call `responseContentAvailable()` -- this was the case for HTTP/2 in particular, where content is notified asynchronously, rather than being created by a call to `ContentSource.read()`.

Signed-off-by: Simone Bordet <[email protected]>

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/transport/HttpReceiver.java

@@ -315,13 +315,35 @@ protected void responseHeaders(HttpExchange exchange)
      * Method to be invoked when response content is available to be read.
      * <p>
      * This method takes care of ensuring the {@link Content.Source} passed to
-     * {@link Response.ContentSourceListener#onContentSource(Response, Content.Source)} calls the
-     * demand callback.
+     * {@link Response.ContentSourceListener#onContentSource(Response, Content.Source)}
+     * calls the demand callback.
+     * The call to the demand callback is serialized with other events.
      */
-    protected void responseContentAvailable()
+    protected void responseContentAvailable(HttpExchange exchange)
     {
         if (LOG.isDebugEnabled())
-            LOG.debug("Response content available on {}", this);
+            LOG.debug("Invoking responseContentAvailable on {}", this);
+
+        invoker.run(() ->
+        {
+            if (LOG.isDebugEnabled())
+                LOG.debug("Executing responseContentAvailable on {}", this);
+
+            if (exchange.isResponseCompleteOrTerminated())
+                return;
+
+            responseContentAvailable();
+        });
+    }
+
+    /**
+     * Method to be invoked when response content is available to be read.
+     * <p>
+     * This method directly invokes the demand callback, assuming the caller
+     * is already serialized with other events.
+     */
+    protected void responseContentAvailable()
+    {
         contentSource.onDataAvailable();
     }
 
@@ -712,9 +734,9 @@ public void onDataAvailable()
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("onDataAvailable on {}", this);
-            // The demandCallback will call read() that will itself call
-            // HttpReceiver.read(boolean) so it must be called by the invoker.
-            invokeDemandCallback(true);
+            // The onDataAvailable() method is only ever called
+            // by the invoker so avoid using the invoker again.
+            invokeDemandCallback(false);
         }
 
         @Override
@@ -760,29 +782,28 @@ private void processDemand()
                 }
             }
 
-            // The processDemand method is only ever called by the
-            // invoker so there is no need to use the latter here.
+            // The processDemand() method is only ever called
+            // by the invoker so avoid using the invoker again.
             invokeDemandCallback(false);
         }
 
         private void invokeDemandCallback(boolean invoke)
         {
             Runnable demandCallback = demandCallbackRef.getAndSet(null);
             if (LOG.isDebugEnabled())
-                LOG.debug("Invoking demand callback on {}", this);
-            if (demandCallback != null)
+                LOG.debug("Invoking demand callback {} on {}", demandCallback, this);
+            if (demandCallback == null)
+                return;
+            try
             {
-                try
-                {
-                    if (invoke)
-                        invoker.run(demandCallback);
-                    else
-                        demandCallback.run();
-                }
-                catch (Throwable x)
-                {
-                    fail(x);
-                }
+                if (invoke)
+                    invoker.run(demandCallback);
+                else
+                    demandCallback.run();
+            }
+            catch (Throwable x)
+            {
+                fail(x);
             }
         }
 

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/transport/internal/HttpReceiverOverHTTP.java

@@ -79,7 +79,9 @@ void receive()
         }
         else
         {
-            responseContentAvailable();
+            HttpExchange exchange = getHttpExchange();
+            if (exchange != null)
+                responseContentAvailable(exchange);
         }
     }
 

jetty-core/jetty-fcgi/jetty-fcgi-client/src/main/java/org/eclipse/jetty/fcgi/client/transport/internal/HttpReceiverOverFCGI.java

@@ -40,7 +40,9 @@ void receive()
         }
         else
         {
-            responseContentAvailable();
+            HttpExchange exchange = getHttpExchange();
+            if (exchange != null)
+                responseContentAvailable(exchange);
         }
     }
 
@@ -107,6 +109,9 @@ public void failAndClose(Throwable failure)
 
     void content(Content.Chunk chunk)
     {
+        HttpExchange exchange = getHttpExchange();
+        if (exchange == null)
+            return;
         if (this.chunk != null)
             throw new IllegalStateException();
         // Retain the chunk because it is stored for later reads.

jetty-core/jetty-http2/jetty-http2-client-transport/src/main/java/org/eclipse/jetty/http2/client/transport/internal/HttpReceiverOverHTTP2.java

@@ -50,8 +50,6 @@ public class HttpReceiverOverHTTP2 extends HttpReceiver implements HTTP2Channel.
 {
     private static final Logger LOG = LoggerFactory.getLogger(HttpReceiverOverHTTP2.class);
 
-    private final Runnable onDataAvailableTask = new Invocable.ReadyTask(Invocable.InvocationType.NON_BLOCKING, this::responseContentAvailable);
-
     public HttpReceiverOverHTTP2(HttpChannel channel)
     {
         super(channel);
@@ -213,7 +211,7 @@ public Runnable onDataAvailable()
         HttpExchange exchange = getHttpExchange();
         if (exchange == null)
             return null;
-        return onDataAvailableTask;
+        return new Invocable.ReadyTask(Invocable.InvocationType.NON_BLOCKING, () -> responseContentAvailable(exchange));
     }
 
     @Override

jetty-core/jetty-http3/jetty-http3-client-transport/src/main/java/org/eclipse/jetty/http3/client/transport/internal/HttpReceiverOverHTTP3.java

@@ -127,7 +127,7 @@ public void onDataAvailable(Stream.Client stream)
         if (exchange == null)
             return;
 
-        responseContentAvailable();
+        responseContentAvailable(exchange);
     }
 
     @Override