Fixes #12652 - Jetty Reactive client hangs for HTTP 401 responses.

Fixed ResponseListeners.emitEvents() to emit the "contentSource" event in all cases.

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

Author: sbordet

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/ContentDecoder.java

@@ -13,6 +13,7 @@
 
 package org.eclipse.jetty.client;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
@@ -21,6 +22,7 @@
 import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.io.RetainableByteBuffer;
+import org.eclipse.jetty.util.component.Dumpable;
 
 /**
  * {@link ContentDecoder} decodes content bytes of a response.
@@ -109,7 +111,7 @@ public int hashCode()
         public abstract ContentDecoder newContentDecoder();
     }
 
-    public static class Factories implements Iterable<ContentDecoder.Factory>
+    public static class Factories implements Iterable<ContentDecoder.Factory>, Dumpable
     {
         private final Map<String, Factory> factories = new LinkedHashMap<>();
         private HttpField acceptEncodingField;
@@ -138,5 +140,17 @@ public Factory put(Factory factory)
             acceptEncodingField = new HttpField(HttpHeader.ACCEPT_ENCODING, value);
             return result;
         }
+
+        @Override
+        public String dump()
+        {
+            return Dumpable.dump(this);
+        }
+
+        @Override
+        public void dump(Appendable out, String indent) throws IOException
+        {
+            Dumpable.dumpObjects(out, indent, this, factories);
+        }
     }
 }

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

@@ -27,6 +27,7 @@
 import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.io.Content;
 import org.eclipse.jetty.io.content.ByteBufferContentSource;
+import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.ExceptionUtil;
 import org.eclipse.jetty.util.thread.AutoLock;
 import org.eclipse.jetty.util.thread.Invocable;
@@ -392,15 +393,14 @@ private void emitEvents(Response response)
                 iterator.remove();
         }
         notifyHeaders(headersListener, response);
+        ByteBuffer content = BufferUtil.EMPTY_BUFFER;
         if (response instanceof ContentResponse contentResponse)
         {
-            byte[] content = contentResponse.getContent();
-            if (content != null && content.length > 0)
-            {
-                ByteBufferContentSource byteBufferContentSource = new ByteBufferContentSource(ByteBuffer.wrap(content));
-                notifyContentSource(contentSourceListener, response, byteBufferContentSource);
-            }
+            byte[] bytes = contentResponse.getContent();
+            if (bytes != null && bytes.length > 0)
+                content = ByteBuffer.wrap(bytes);
         }
+        notifyContentSource(contentSourceListener, response, new ByteBufferContentSource(content));
     }
 
     public void emitSuccess(Response response)

jetty-core/jetty-client/src/test/java/org/eclipse/jetty/client/transport/ResponseListenersTest.java

@@ -15,14 +15,19 @@
 
 import java.io.Closeable;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeoutException;
 
 import org.eclipse.jetty.client.Response;
+import org.eclipse.jetty.client.internal.HttpContentResponse;
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.http.HttpFields;
 import org.eclipse.jetty.io.Content;
 import org.eclipse.jetty.io.content.ChunksContentSource;
+import org.eclipse.jetty.util.BufferUtil;
 import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -227,6 +232,52 @@ public void run()
         contentSource.close();
     }
 
+    @Test
+    public void testEmitEventsInvokesContentSourceListenerForNoContent()
+    {
+        ResponseListeners responseListeners = new ResponseListeners();
+        List<String> events = new ArrayList<>();
+        responseListeners.addListener(new Response.Listener()
+        {
+            @Override
+            public void onBegin(Response response)
+            {
+                events.add("BEGIN");
+            }
+
+            @Override
+            public boolean onHeader(Response response, HttpField field)
+            {
+                return events.add("HEADER");
+            }
+
+            @Override
+            public void onHeaders(Response response)
+            {
+                events.add("HEADERS");
+            }
+
+            @Override
+            public void onContentSource(Response response, Content.Source contentSource)
+            {
+                events.add("CONTENT-SOURCE");
+            }
+
+            @Override
+            public void onSuccess(Response response)
+            {
+                events.add("SUCCESS");
+            }
+        });
+
+        Response response = new HttpResponse(null).addHeader(HttpFields.CONTENT_LENGTH_0);
+        Response contentResponse = new HttpContentResponse(response, BufferUtil.EMPTY_BYTES, null, null);
+        responseListeners.emitSuccess(contentResponse);
+
+        List<String> expected = List.of("BEGIN", "HEADER", "HEADERS", "CONTENT-SOURCE", "SUCCESS");
+        assertThat(events, is(expected));
+    }
+
     private static class TestSource extends ChunksContentSource implements Closeable
     {
         private Content.Chunk[] chunks;