Revert "deadlock fix v2"

This reverts commit 61c273b.

sdk/clientcore/core/src/test/java/io/clientcore/core/shared/LocalTestServer.java

@@ -60,41 +60,45 @@ public LocalTestServer(HttpProtocolVersion supportedProtocol, boolean includeTls
             httpConfig.addCustomizer(new SecureRequestCustomizer());
         }
 
-        final ServerConnector httpConnector;
-        if (!includeTls) {
-            HttpConnectionFactory http11 = new HttpConnectionFactory(httpConfig);
-            httpConnector = new ServerConnector(server, http11);
-        } else {
+        List<ConnectionFactory> connectionFactories = new ArrayList<>();
+
+        // SSL/TLS connection factory
+        if (includeTls) {
+            String nextProtocol = supportedProtocol == null
+                ? HttpVersion.HTTP_1_1.asString()
+                : (supportedProtocol == HttpProtocolVersion.HTTP_1_1) ? HttpVersion.HTTP_1_1.asString() : "alpn";
+
             Security.addProvider(new OpenSSLProvider());
             SslContextFactory.Server sslContextFactory = new SslContextFactory.Server();
             sslContextFactory.setProvider("Conscrypt");
             String mockKeyStore = Objects.toString(LocalTestServer.class.getResource("/keystore.jks"), null);
             sslContextFactory.setKeyStorePath(mockKeyStore);
             sslContextFactory.setKeyStorePassword("password");
             sslContextFactory.setKeyManagerPassword("password");
+            sslContextFactory.setKeyStorePath(mockKeyStore);
             sslContextFactory.setTrustStorePassword("password");
             sslContextFactory.setTrustAll(true);
 
-            sslContextFactory.setCipherComparator(java.util.Comparator.comparingInt(String::hashCode));
-
-            SslConnectionFactory ssl = new SslConnectionFactory(sslContextFactory, "alpn");
-
-            HttpConnectionFactory http11 = new HttpConnectionFactory(httpConfig);
+            connectionFactories.add(new SslConnectionFactory(sslContextFactory, nextProtocol));
+        }
 
-            if (supportedProtocol == HttpProtocolVersion.HTTP_2) {
-                HTTP2ServerConnectionFactory http2 = new HTTP2ServerConnectionFactory(httpConfig);
-                ALPNServerConnectionFactory alpn = new ALPNServerConnectionFactory();
-                alpn.setDefaultProtocol(http11.getProtocol());
+        if (supportedProtocol == HttpProtocolVersion.HTTP_2) {
+            // ALPN connection factory
+            // HTTP/2 connection factory
+            ALPNServerConnectionFactory alpn = new ALPNServerConnectionFactory();
+            alpn.setDefaultProtocol("h2");
+            connectionFactories.add(alpn);
+            connectionFactories.add(new HTTP2ServerConnectionFactory(httpConfig));
+        }
 
-                httpConnector = new ServerConnector(server, ssl, alpn, http2, http11);
-            } else {
-                httpConnector = new ServerConnector(server, ssl, http11);
-            }
+        if (supportedProtocol == null || supportedProtocol == HttpProtocolVersion.HTTP_1_1) {
+            // HTTP/1.1 connection factory
+            connectionFactories.add(new HttpConnectionFactory(httpConfig));
         }
 
-        this.connector = httpConnector;
-        this.connector.setHost("localhost");
-        server.addConnector(this.connector);
+        connector = new ServerConnector(server, connectionFactories.toArray(new ConnectionFactory[0]));
+        connector.setHost("localhost");
+        server.addConnector(connector);
 
         ServletContextHandler servletContextHandler = new ServletContextHandler();
         servletContextHandler.setContextPath("/");

sdk/clientcore/http-netty4/spotbugs-exclude.xml

@@ -62,8 +62,4 @@
     <Bug pattern="VO_VOLATILE_REFERENCE_TO_ARRAY" />
     <Class name="io.clientcore.http.netty4.implementation.Netty4ChannelBinaryData" />
   </Match>
-  <Match>
-    <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />
-    <Class name="io.clientcore.http.netty4.implementation.Netty4ChannelInputStream" />
-  </Match>
 </FindBugsFilter>

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/NettyHttpClient.java

@@ -156,26 +156,19 @@ public Response<BinaryData> send(HttpRequest request) {
             }
 
             response = new Response<>(request, info.getStatusCode(), info.getHeaders(), body);
-            Channel channel = info.getResponseChannel();
-            if (channel != null) {
-                cleanupPipeline(channel.pipeline());
-            }
         } else {
             // Otherwise we aren't finished, handle the remaining content according to the documentation in
             // 'channelRead()'.
             BinaryData body = BinaryData.empty();
             ResponseBodyHandling bodyHandling = info.getResponseBodyHandling();
             Channel channel = info.getResponseChannel();
-            ChannelPipeline pipeline = channel.pipeline();
             if (bodyHandling == ResponseBodyHandling.IGNORE) {
                 // We're ignoring the response content.
                 CountDownLatch drainLatch = new CountDownLatch(1);
                 channel.pipeline().addLast(new Netty4EagerConsumeChannelHandler(drainLatch, ignored -> {
                 }, info.isHttp2()));
                 channel.config().setAutoRead(true);
                 awaitLatch(drainLatch);
-
-                cleanupPipeline(pipeline);
             } else if (bodyHandling == ResponseBodyHandling.STREAM) {
                 // Body streaming uses a special BinaryData that tracks the firstContent read and the Channel it came
                 // from so it can be consumed when the BinaryData is being used.
@@ -205,8 +198,6 @@ public Response<BinaryData> send(HttpRequest request) {
                 channel.config().setAutoRead(true);
                 awaitLatch(drainLatch);
 
-                cleanupPipeline(pipeline);
-
                 body = BinaryData.fromBytes(info.getEagerContent().toByteArray());
             }
 
@@ -362,11 +353,4 @@ private Netty4ConnectionPoolKey constructConnectionPoolKey(SocketAddress finalDe
         return key;
     }
 
-    private void cleanupPipeline(ChannelPipeline pipeline) {
-        Netty4PipelineCleanupHandler pipelineCleanupHandler = pipeline.get(Netty4PipelineCleanupHandler.class);
-        if (pipelineCleanupHandler != null) {
-            pipelineCleanupHandler.cleanup(pipeline.context(pipelineCleanupHandler), false);
-        }
-    }
-
 }

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4ChannelBinaryData.java

@@ -213,11 +213,6 @@ private void drainStream() {
                     throw CoreException.from(exception);
                 }
             }
-
-            Netty4PipelineCleanupHandler cleanupHandler = channel.pipeline().get(Netty4PipelineCleanupHandler.class);
-            if (cleanupHandler != null) {
-                cleanupHandler.cleanup(channel.pipeline().context(cleanupHandler), false);
-            }
         }
     }
 }

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4ChannelInputStream.java

@@ -7,10 +7,8 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.LinkedList;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Implementation of {@link InputStream} that reads contents from a Netty {@link Channel}.
@@ -20,9 +18,6 @@ public final class Netty4ChannelInputStream extends InputStream {
     private final boolean isHttp2;
     private final Runnable onClose;
 
-    //Ensures the close operation is idempotent.
-    private final AtomicBoolean closed = new AtomicBoolean(false);
-
     // Indicator for the Channel being fully read.
     // This will become true before 'streamDone' becomes true, but both may become true in the same operation.
     // Once this is true, the Channel will never be read again.
@@ -33,9 +28,9 @@ public final class Netty4ChannelInputStream extends InputStream {
     // Once this is true, the stream will never return data again.
     private boolean streamDone = false;
 
-    // Queue of byte[]s that maintains the last available contents from the Channel / eager content.
-    // A queue is needed as each Channel.read() may result in many channelRead calls.
-    private final Queue<byte[]> additionalBuffers;
+    // Linked list of byte[]s that maintains the last available contents from the Channel / eager content.
+    // A list is needed as each Channel.read() may result in many channelRead calls.
+    private final LinkedList<byte[]> additionalBuffers;
 
     private byte[] currentBuffer;
 
@@ -62,7 +57,7 @@ public final class Netty4ChannelInputStream extends InputStream {
             this.currentBuffer = new byte[0];
         }
         this.readIndex = 0;
-        this.additionalBuffers = new ConcurrentLinkedQueue<>();
+        this.additionalBuffers = new LinkedList<>();
         this.channel = channel;
         if (channel.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
             channel.pipeline().remove(Netty4InitiateOneReadHandler.class);
@@ -183,21 +178,19 @@ public long skip(long n) throws IOException {
      */
     @Override
     public void close() throws IOException {
-        if (closed.compareAndSet(false, true)) {
-            try {
-                if (onClose != null) {
-                    onClose.run();
-                }
-            } finally {
-                super.close();
-                streamDone = true;
+        try {
+            if (onClose != null && !streamDone) {
+                onClose.run();
             }
+        } finally {
+            super.close();
+            streamDone = true;
         }
     }
 
     private boolean setupNextBuffer() throws IOException {
         if (!additionalBuffers.isEmpty()) {
-            currentBuffer = additionalBuffers.poll();
+            currentBuffer = additionalBuffers.pop();
             readIndex = 0;
             return true;
         } else if (readMore()) {
@@ -232,7 +225,7 @@ private boolean readMore() throws IOException {
                 byte[] buffer = new byte[byteBuf.readableBytes()];
                 byteBuf.readBytes(buffer);
 
-                additionalBuffers.offer(buffer);
+                additionalBuffers.add(buffer);
             }, isHttp2);
             channel.pipeline().addLast(Netty4HandlerNames.READ_ONE, handler);
         }
@@ -260,7 +253,7 @@ private boolean readMore() throws IOException {
         }
 
         if (!additionalBuffers.isEmpty()) {
-            currentBuffer = additionalBuffers.poll();
+            currentBuffer = additionalBuffers.pop();
             readIndex = 0;
         } else if (channelDone) { // Don't listen to IntelliJ here, channelDone may be false.
             // This read contained no data and the channel completed, therefore the stream is also completed.

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4EagerConsumeChannelHandler.java

@@ -57,38 +57,28 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
             } else {
                 lastRead = msg instanceof LastHttpContent;
             }
-
-            if (lastRead) {
-                latch.countDown();
-            }
-
             ctx.fireChannelRead(msg);
 
         } catch (IOException | RuntimeException ex) {
             ReferenceCountUtil.release(msg);
-            latch.countDown();
-            Netty4PipelineCleanupHandler cleanupHandler = ctx.pipeline().get(Netty4PipelineCleanupHandler.class);
-            if (cleanupHandler != null) {
-                cleanupHandler.cleanup(ctx, true);
-            } else {
-                ctx.close();
-            }
+            ctx.fireExceptionCaught(ex);
+            cleanup(ctx);
         }
     }
 
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) {
         ctx.fireChannelReadComplete();
         if (lastRead) {
-            latch.countDown();
+            cleanup(ctx);
         }
     }
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         this.exception = cause;
-        latch.countDown();
         ctx.fireExceptionCaught(cause);
+        cleanup(ctx);
     }
 
     Throwable channelException() {
@@ -98,13 +88,13 @@ Throwable channelException() {
     // TODO (alzimmer): Are the latch countdowns needed for unregistering and inactivity?
     @Override
     public void channelUnregistered(ChannelHandlerContext ctx) {
-        latch.countDown();
+        cleanup(ctx);
         ctx.fireChannelUnregistered();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) {
-        latch.countDown();
+        cleanup(ctx);
         ctx.fireChannelInactive();
     }
 
@@ -115,7 +105,15 @@ public void handlerAdded(ChannelHandlerContext ctx) {
             // an exception. Simply counting down the latch would cause the caller to receive
             // an empty/incomplete data stream without any indication of the underlying network error.
             ctx.fireExceptionCaught(new ClosedChannelException());
+            ctx.pipeline().remove(this);
         }
     }
 
+    private void cleanup(ChannelHandlerContext ctx) {
+        if (ctx.pipeline().get(Netty4EagerConsumeChannelHandler.class) != null) {
+            ctx.pipeline().remove(this);
+        }
+
+        latch.countDown();
+    }
 }

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4InitiateOneReadHandler.java

@@ -95,6 +95,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) {
         latch.countDown();
+        if (lastRead) {
+            ctx.pipeline().remove(this);
+        }
         ctx.fireChannelReadComplete();
     }
 
@@ -106,6 +109,9 @@ boolean isChannelConsumed() {
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         this.exception = cause;
         latch.countDown();
+        if (ctx.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
+            ctx.pipeline().remove(this);
+        }
         ctx.fireExceptionCaught(cause);
     }
 
@@ -117,12 +123,18 @@ Throwable channelException() {
     @Override
     public void channelUnregistered(ChannelHandlerContext ctx) {
         latch.countDown();
+        if (ctx.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
+            ctx.pipeline().remove(this);
+        }
         ctx.fireChannelUnregistered();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) {
         latch.countDown();
+        if (ctx.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
+            ctx.pipeline().remove(this);
+        }
         ctx.fireChannelInactive();
     }
 
@@ -133,6 +145,7 @@ public void handlerAdded(ChannelHandlerContext ctx) {
             // an exception. Simply counting down the latch would cause the caller to receive
             // an empty/incomplete data stream without any indication of the underlying network error.
             ctx.fireExceptionCaught(new ClosedChannelException());
+            ctx.pipeline().remove(this);
         }
     }
 }