Add Unix domain socket support

This change adds Unix domain socket support. The sync client uses
JUnixSocket, which provides synchronous UDS support through the legacy
`java.net.Socket` API; the async client uses the JEP 380 implementation
of UDS through the `java.nio.channels.SocketChannel` API (requires JDK16
or later).

Since the synchronous client is tightly coupled to the `Socket` API, we
can't trivially use JEP 380 UDS support. We would first have to write an
adapter to implement the `Socket` API, backed by a JEP 380 UDS
`SocketChannel`. This would require us to implement features like socket
option configuration, connection timeouts, and socket timeouts; we would
also have to implement APIs like `getInetAddress()` which don't actually
make sense in a UDS context. This is probably doable (JUnixSocket does
it, albeit with a different implementation strategy based on native
code), but it's not trivial.

The asynchronous client is the other way around: it supports JEP 380,
but not JUnixSocket. The issue here is more subtle: JDK and JUnixDomain
channels cannot be mixed in the same selector, and since JUnixDomain
does not provide an implementation of TCP/IP channels, supporting
JUnixSocket in the async client would require substantial rework in the
IO reactor. Since JDK8 is end-of-life next year, I doubt this is worth
doing unless we can find some clever way of integrating the new channel
type with minimal churn.

Unix domain socket support is exposed through the `RequestConfig` API. A
path to a Unix domain socket can be provided as a client-wide default
through `setDefaultRequestConfig`, or on a per-request basis through
`setConfig`. Currently, proxies and TLS are not supported through UDS.
The former feature seems unnecessary, but the latter is likely worth
adding at some point, since contacting an HTTPS endpoint over UDS
(sometimes denoted by the URI scheme `https+unix`) is not unheard of.

Author: rschmitt

httpclient5-testing/pom.xml

@@ -77,6 +77,12 @@
       <artifactId>httpclient5-fluent</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.kohlschutter.junixsocket</groupId>
+      <artifactId>junixsocket-core</artifactId>
+      <scope>test</scope>
+      <type>pom</type>
+    </dependency>
     <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter-params</artifactId>
@@ -153,4 +159,4 @@
     </plugins>
   </reporting>
 
-</project>
+</project>

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/sync/StandardTestClientBuilder.java

@@ -27,6 +27,7 @@
 
 package org.apache.hc.client5.testing.extension.sync;
 
+import java.nio.file.Path;
 import java.util.Collection;
 
 import org.apache.hc.client5.http.AuthenticationStrategy;
@@ -36,6 +37,7 @@
 import org.apache.hc.client5.http.auth.CredentialsProvider;
 import org.apache.hc.client5.http.classic.ExecChainHandler;
 import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.config.RequestConfig;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
 import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
@@ -157,6 +159,14 @@ public TestClientBuilder setDefaultCredentialsProvider(final CredentialsProvider
         return this;
     }
 
+    @Override
+    public TestClientBuilder setUnixDomainSocket(final Path unixDomainSocket) {
+        this.clientBuilder.setDefaultRequestConfig(RequestConfig.custom()
+                .setUnixDomainSocket(unixDomainSocket)
+                .build());
+        return this;
+    }
+
     @Override
     public TestClient build() throws Exception {
         final HttpClientConnectionManager connectionManagerCopy = connectionManager != null ? connectionManager :

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/sync/TestClientBuilder.java

@@ -27,6 +27,7 @@
 
 package org.apache.hc.client5.testing.extension.sync;
 
+import java.nio.file.Path;
 import java.util.Collection;
 
 import org.apache.hc.client5.http.AuthenticationStrategy;
@@ -103,6 +104,10 @@ default TestClientBuilder setDefaultCredentialsProvider(CredentialsProvider cred
         throw new UnsupportedOperationException("Operation not supported by " + getProtocolLevel());
     }
 
+    default TestClientBuilder setUnixDomainSocket(Path unixDomainSocket) {
+        throw new UnsupportedOperationException("Operation not supported by " + getProtocolLevel());
+    }
+
     TestClient build() throws Exception;
 
 }

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/sync/TestClientResources.java

@@ -50,6 +50,7 @@ public class TestClientResources implements AfterEachCallback {
 
     private TestServer server;
     private TestClient client;
+    private UnixDomainProxyServer udsProxy;
 
     public TestClientResources(final URIScheme scheme, final ClientProtocolLevel clientProtocolLevel, final Timeout timeout) {
         this.scheme = scheme != null ? scheme : URIScheme.HTTP;
@@ -74,6 +75,9 @@ public void afterEach(final ExtensionContext extensionContext) {
         if (client != null) {
             client.close(CloseMode.GRACEFUL);
         }
+        if (udsProxy != null) {
+            udsProxy.close();
+        }
         if (server != null) {
             server.shutdown(CloseMode.IMMEDIATE);
         }
@@ -99,6 +103,15 @@ public TestServer server() throws Exception {
         return server;
     }
 
+    public UnixDomainProxyServer udsProxy() throws Exception {
+        if (udsProxy == null) {
+            final TestServer testServer = server();
+            final int port = testServer.getServerAddress().getPort();
+            udsProxy = new UnixDomainProxyServer(port);
+        }
+        return udsProxy;
+    }
+
     public void configureClient(final Consumer<TestClientBuilder> clientCustomizer) {
         Asserts.check(client == null, "Client is already running and cannot be changed");
         clientCustomizer.accept(clientBuilder);

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/sync/TestServer.java

@@ -43,6 +43,7 @@ public class TestServer {
     private final Http1Config http1Config;
     private final HttpProcessor httpProcessor;
     private final Decorator<HttpServerRequestHandler> exchangeHandlerDecorator;
+    private volatile InetSocketAddress serverAddress;
 
     TestServer(
             final ClassicTestServer server,
@@ -64,7 +65,14 @@ public InetSocketAddress start() throws IOException {
         server.configure(exchangeHandlerDecorator);
         server.configure(httpProcessor);
         server.start();
-        return new InetSocketAddress(server.getInetAddress(), server.getPort());
+        serverAddress = new InetSocketAddress(server.getInetAddress(), server.getPort());
+        return serverAddress;
     }
 
+    public InetSocketAddress getServerAddress() {
+        if (serverAddress == null) {
+            throw new IllegalStateException("Server has not been started");
+        }
+        return serverAddress;
+    }
 }

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/sync/UnixDomainProxyServer.java

@@ -0,0 +1,137 @@
+/*
+ * ====================================================================
+ * 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.client5.testing.extension.sync;
+
+import org.newsclub.net.unix.AFUNIXServerSocket;
+import org.newsclub.net.unix.AFUNIXSocket;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+
+public final class UnixDomainProxyServer {
+    private final int port;
+    private final ExecutorService executorService;
+    private final Path socketPath;
+    private final CountDownLatch serverReady = new CountDownLatch(1);
+    private volatile AFUNIXServerSocket serverSocket;
+
+    public UnixDomainProxyServer(final int port) {
+        this.port = port;
+        this.executorService = Executors.newCachedThreadPool();
+        this.socketPath = (new File("proxy.sock")).toPath();
+    }
+
+    public void start() {
+        executorService.submit(this::runUdsProxy);
+        try {
+            serverReady.await();
+        } catch (final InterruptedException ex) {
+            throw new RuntimeException(ex);
+        }
+    }
+
+    public Path getSocketPath() {
+        return socketPath;
+    }
+
+    public void close() {
+        try {
+            serverSocket.close();
+            executorService.shutdownNow();
+            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+                throw new RuntimeException("Failed to shut down");
+            }
+            Files.deleteIfExists(socketPath);
+        } catch (final InterruptedException | IOException ex) {
+            throw new RuntimeException(ex);
+        }
+    }
+
+    private void runUdsProxy() {
+        try {
+            Files.deleteIfExists(socketPath);
+        } catch (final IOException ignore) {
+        }
+
+        try {
+            try (final AFUNIXServerSocket server = AFUNIXServerSocket.bindOn(socketPath, true)) {
+                this.serverSocket = server;
+                serverReady.countDown();
+                serveRequests(server);
+            } catch (final Throwable ignore) {
+            }
+        } catch (final Throwable t) {
+            serverReady.countDown();
+            throw t;
+        }
+    }
+
+    private void serveRequests(final AFUNIXServerSocket server) throws IOException {
+        while (true) {
+            final AFUNIXSocket udsClient = server.accept();
+            final Socket tcpSocket = new Socket("localhost", port);
+            final CompletableFuture<Void> f1 = supplyAsync(() -> pipe(udsClient, tcpSocket), executorService);
+            final CompletableFuture<Void> f2 = supplyAsync(() -> pipe(tcpSocket, udsClient), executorService);
+            CompletableFuture.allOf(f1, f2).whenComplete((result, ex) -> {
+                try {
+                    udsClient.close();
+                    tcpSocket.close();
+                } catch (final IOException ignore) {
+                }
+            });
+        }
+    }
+
+    private Void pipe(final Socket inputSocket, final Socket outputSocket) {
+        try (
+            final InputStream in = inputSocket.getInputStream();
+            final OutputStream out = outputSocket.getOutputStream()
+        ) {
+            final byte[] buf = new byte[8192];
+            int len;
+            while ((len = in.read(buf)) != -1) {
+                out.write(buf, 0, len);
+                out.flush();
+            }
+        } catch (final IOException ignore) {
+        }
+        return null;
+    }
+}

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

@@ -28,6 +28,7 @@
 package org.apache.hc.client5.testing.sync;
 
 import java.net.InetSocketAddress;
+import java.nio.file.Path;
 import java.util.function.Consumer;
 
 import org.apache.hc.client5.testing.extension.sync.ClientProtocolLevel;
@@ -47,9 +48,16 @@ abstract class AbstractIntegrationTestBase {
 
     @RegisterExtension
     private final TestClientResources testResources;
+    private final boolean useUnixDomainSocket;
 
     protected AbstractIntegrationTestBase(final URIScheme scheme, final ClientProtocolLevel clientProtocolLevel) {
+        this(scheme, clientProtocolLevel, false);
+    }
+
+    protected AbstractIntegrationTestBase(
+        final URIScheme scheme, final ClientProtocolLevel clientProtocolLevel, final boolean useUnixDomainSocket) {
         this.testResources = new TestClientResources(scheme, clientProtocolLevel, TIMEOUT);
+        this.useUnixDomainSocket = useUnixDomainSocket;
     }
 
     public URIScheme scheme() {
@@ -67,6 +75,9 @@ public void configureServer(final Consumer<TestServerBootstrap> serverCustomizer
     public HttpHost startServer() throws Exception {
         final TestServer server = testResources.server();
         final InetSocketAddress inetSocketAddress = server.start();
+        if (useUnixDomainSocket) {
+            testResources.udsProxy().start();
+        }
         return new HttpHost(testResources.scheme().id, "localhost", inetSocketAddress.getPort());
     }
 
@@ -75,6 +86,12 @@ public void configureClient(final Consumer<TestClientBuilder> clientCustomizer)
     }
 
     public TestClient client() throws Exception {
+        if (useUnixDomainSocket) {
+            final Path socketPath = testResources.udsProxy().getSocketPath();
+            testResources.configureClient(builder -> {
+                builder.setUnixDomainSocket(socketPath);
+            });
+        }
         return testResources.client();
     }