Add socket timeout integration tests

This change adds integration test coverage for various types of read
timeouts. The test coverage includes HTTP, HTTPS, and HTTP over UDS if
available. The `/random` request handlers have been augmented with delay
support, so that the clients can read the response headers and then time
out while reading the entity. The tests make use of `ConnectionConfig`,
`ResponseTimeout`, and (on the sync client only) `SocketConfig`, and
demonstrate the precedence among these config options when more than one
is supplied.

One issue uncovered by these tests is that graceful closure of the async
client does not work when UDS socket timeouts have fired. To work around
this until the issue can be root caused, the async UDS tests use
`CloseMode.IMMEDIATE`.

Author: rschmitt

httpclient5-testing/src/main/java/org/apache/hc/client5/testing/async/AsyncRandomHandler.java

@@ -93,6 +93,20 @@ public void handleRequest(
         } catch (final URISyntaxException ex) {
             throw new ProtocolException(ex.getMessage(), ex);
         }
+        final String query = uri.getQuery();
+        int delayMs = 0;
+        boolean drip = false;
+        if (query != null) {
+            final String[] params = query.split("&");
+            for (final String param : params) {
+                if (param.startsWith("delay=")) {
+                    delayMs = Integer.parseInt(param.substring(1 + param.indexOf('=')));
+                } else if (param.equals("drip=1")) {
+                    drip = true;
+                }
+            }
+        }
+
         final String path = uri.getPath();
         final int slash = path.lastIndexOf('/');
         if (slash != -1) {
@@ -109,7 +123,7 @@ public void handleRequest(
                 n = 1 + (int)(Math.random() * 79.0);
             }
             final HttpResponse response = new BasicHttpResponse(HttpStatus.SC_OK);
-            final AsyncEntityProducer entityProducer = new RandomBinAsyncEntityProducer(n);
+            final AsyncEntityProducer entityProducer = new RandomBinAsyncEntityProducer(n, delayMs, drip);
             entityProducerRef.set(entityProducer);
             responseChannel.sendResponse(response, entityProducer, context);
         } else {
@@ -162,12 +176,22 @@ public static class RandomBinAsyncEntityProducer extends AbstractBinAsyncEntityP
         private final long length;
         private long remaining;
         private final ByteBuffer buffer;
+        private final int delayMs;
+        private final boolean drip;
+        private volatile long deadline;
 
         public RandomBinAsyncEntityProducer(final long len) {
+            this(len, 0, false);
+        }
+
+        public RandomBinAsyncEntityProducer(final long len, final int delayMs, final boolean drip) {
             super(512, ContentType.DEFAULT_TEXT);
             length = len;
             remaining = len;
             buffer = ByteBuffer.allocate(1024);
+            this.delayMs = delayMs;
+            this.deadline = System.currentTimeMillis() + (drip ? 0 : delayMs);
+            this.drip = drip;
         }
 
         @Override
@@ -192,6 +216,10 @@ public int availableData() {
 
         @Override
         protected void produceData(final StreamChannel<ByteBuffer> channel) throws IOException {
+            if (System.currentTimeMillis() < deadline) {
+                return;
+            }
+
             final int chunk = Math.min((int) (remaining < Integer.MAX_VALUE ? remaining : Integer.MAX_VALUE), buffer.remaining());
             for (int i = 0; i < chunk; i++) {
                 final byte b = RANGE[(int) (Math.random() * RANGE.length)];
@@ -205,6 +233,8 @@ protected void produceData(final StreamChannel<ByteBuffer> channel) throws IOExc
 
             if (remaining <= 0 && buffer.position() == 0) {
                 channel.endStream();
+            } else if (drip) {
+                deadline = System.currentTimeMillis() + delayMs;
             }
         }
 

httpclient5-testing/src/main/java/org/apache/hc/client5/testing/classic/RandomHandler.java

@@ -84,6 +84,20 @@ public void handle(final ClassicHttpRequest request,
         } catch (final URISyntaxException ex) {
             throw new ProtocolException(ex.getMessage(), ex);
         }
+        final String query = uri.getQuery();
+        int delayMs = 0;
+        boolean drip = false;
+        if (query != null) {
+            final String[] params = query.split("&");
+            for (final String param : params) {
+                if (param.startsWith("delay=")) {
+                    delayMs = Integer.parseInt(param.substring(1 + param.indexOf('=')));
+                } else if (param.equals("drip=1")) {
+                    drip = true;
+                }
+            }
+        }
+
         final String path = uri.getPath();
         final int slash = path.lastIndexOf('/');
         if (slash != -1) {
@@ -100,7 +114,7 @@ public void handle(final ClassicHttpRequest request,
                 n = 1 + (int)(Math.random() * 79.0);
             }
             response.setCode(HttpStatus.SC_OK);
-            response.setEntity(new RandomEntity(n));
+            response.setEntity(new RandomEntity(n, delayMs, drip));
         } else {
             throw new ProtocolException("Invalid request path: " + path);
         }
@@ -120,6 +134,12 @@ public static class RandomEntity extends AbstractHttpEntity {
         /** The length of the random data to generate. */
         protected final long length;
 
+        /** The duration of the delay before sending the response entity. */
+        protected final int delayMs;
+
+        /** Whether to delay after each chunk sent. If {@code false},
+         * will only delay at the start of the response entity. */
+        protected final boolean drip;
 
         /**
          * Creates a new entity generating the given amount of data.
@@ -128,8 +148,22 @@ public static class RandomEntity extends AbstractHttpEntity {
          *              0 to maxint
          */
         public RandomEntity(final long len) {
+            this(len, 0, false);
+        }
+
+        /**
+         * Creates a new entity generating the given amount of data.
+         *
+         * @param len       the number of random bytes to generate,
+         *                  0 to maxint
+         * @param delayMs   how long to wait before sending the first byte
+         * @param drip      whether to repeat the delay after each byte
+         */
+        public RandomEntity(final long len, final int delayMs, final boolean drip) {
             super((ContentType) null, null);
-            length = len;
+            this.length = len;
+            this.delayMs = delayMs;
+            this.drip = drip;
         }
 
         /**
@@ -185,31 +219,48 @@ public InputStream getContent() {
         @Override
         public void writeTo(final OutputStream out) throws IOException {
 
-            final int blocksize = 2048;
-            int remaining = (int) length; // range checked in constructor
-            final byte[] data = new byte[Math.min(remaining, blocksize)];
+            try {
+                final int blocksize = 2048;
+                int remaining = (int) length; // range checked in constructor
+                final byte[] data = new byte[Math.min(remaining, blocksize)];
 
-            while (remaining > 0) {
-                final int end = Math.min(remaining, data.length);
+                out.flush();
+                if (!drip) {
+                    delay();
+                }
+                while (remaining > 0) {
+                    final int end = Math.min(remaining, data.length);
 
-                double value = 0.0;
-                for (int i = 0; i < end; i++) {
-                    // we get 5 random characters out of one random value
-                    if (i % 5 == 0) {
-                        value = Math.random();
+                    double value = 0.0;
+                    for (int i = 0; i < end; i++) {
+                        // we get 5 random characters out of one random value
+                        if (i % 5 == 0) {
+                            value = Math.random();
+                        }
+                        value = value * RANGE.length;
+                        final int d = (int) value;
+                        value = value - d;
+                        data[i] = RANGE[d];
                     }
-                    value = value * RANGE.length;
-                    final int d = (int) value;
-                    value = value - d;
-                    data[i] = RANGE[d];
-                }
-                out.write(data, 0, end);
-                out.flush();
+                    out.write(data, 0, end);
+                    out.flush();
 
-                remaining = remaining - end;
+                    remaining = remaining - end;
+                    if (drip) {
+                        delay();
+                    }
+                }
+            } finally {
+                out.close();
             }
-            out.close();
+        }
 
+        private void delay() throws IOException {
+            try {
+                Thread.sleep(delayMs);
+            } catch (final InterruptedException ex) {
+                throw new IOException(ex);
+            }
         }
 
         @Override

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/async/AbstractIntegrationTestBase.java

@@ -28,6 +28,7 @@
 package org.apache.hc.client5.testing.async;
 
 import java.net.InetSocketAddress;
+import java.nio.file.Path;
 import java.util.function.Consumer;
 
 import org.apache.hc.client5.testing.extension.async.ClientProtocolLevel;
@@ -48,9 +49,17 @@ abstract class AbstractIntegrationTestBase {
 
     @RegisterExtension
     private final TestAsyncResources testResources;
+    private final boolean useUnixDomainSocket;
 
     protected AbstractIntegrationTestBase(final URIScheme scheme, final ClientProtocolLevel clientProtocolLevel, final ServerProtocolLevel serverProtocolLevel) {
+        this(scheme, clientProtocolLevel, serverProtocolLevel, false);
+    }
+
+    protected AbstractIntegrationTestBase(
+        final URIScheme scheme, final ClientProtocolLevel clientProtocolLevel,
+        final ServerProtocolLevel serverProtocolLevel, final boolean useUnixDomainSocket) {
         this.testResources = new TestAsyncResources(scheme, clientProtocolLevel, serverProtocolLevel, TIMEOUT);
+        this.useUnixDomainSocket = useUnixDomainSocket;
     }
 
     public URIScheme scheme() {
@@ -72,6 +81,9 @@ public void configureServer(final Consumer<TestAsyncServerBootstrap> serverCusto
     public HttpHost startServer() throws Exception {
         final TestAsyncServer server = testResources.server();
         final InetSocketAddress inetSocketAddress = server.start();
+        if (useUnixDomainSocket) {
+            testResources.udsProxy().start();
+        }
         return new HttpHost(testResources.scheme().id, "localhost", inetSocketAddress.getPort());
     }
 
@@ -80,9 +92,21 @@ public void configureClient(final Consumer<TestAsyncClientBuilder> clientCustomi
     }
 
     public TestAsyncClient startClient() throws Exception {
+        if (useUnixDomainSocket) {
+            final Path socketPath = getUnixDomainSocket();
+            testResources.configureClient(builder -> {
+                builder.setUnixDomainSocket(socketPath);
+            });
+        }
         final TestAsyncClient client = testResources.client();
         client.start();
         return client;
     }
 
+    public Path getUnixDomainSocket() throws Exception {
+        if (useUnixDomainSocket) {
+            return testResources.udsProxy().getSocketPath();
+        }
+        return null;
+    }
 }