HTTPCORE-798: Make socket channel creation customizable

httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java

@@ -28,13 +28,16 @@
 package org.apache.hc.core5.reactor;
 
 import java.io.IOException;
+import java.net.SocketAddress;
+import java.nio.channels.SocketChannel;
 import java.util.concurrent.ThreadFactory;
 
 import org.apache.hc.core5.annotation.Internal;
 import org.apache.hc.core5.concurrent.DefaultThreadFactory;
 import org.apache.hc.core5.function.Callback;
 import org.apache.hc.core5.function.Decorator;
 import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.io.IOFunction;
 import org.apache.hc.core5.util.Args;
 import org.apache.hc.core5.util.TimeValue;
 
@@ -69,6 +72,25 @@ public DefaultConnectingIOReactor(
             final IOReactorMetricsListener threadPoolListener,
             final Callback<IOSession> sessionShutdownCallback,
             final IOWorkerSelector workerSelector) {
+        this(eventHandlerFactory, ioReactorConfig, threadFactory, ioSessionDecorator, exceptionCallback,
+                sessionListener, threadPoolListener, sessionShutdownCallback, workerSelector, null);
+    }
+
+    /**
+     * @since 5.5
+     */
+    @Internal
+    public DefaultConnectingIOReactor(
+            final IOEventHandlerFactory eventHandlerFactory,
+            final IOReactorConfig ioReactorConfig,
+            final ThreadFactory threadFactory,
+            final Decorator<IOSession> ioSessionDecorator,
+            final Callback<Exception> exceptionCallback,
+            final IOSessionListener sessionListener,
+            final IOReactorMetricsListener threadPoolListener,
+            final Callback<IOSession> sessionShutdownCallback,
+            final IOWorkerSelector workerSelector,
+            final IOFunction<SocketAddress, SocketChannel> socketChannelFactory) {
         Args.notNull(eventHandlerFactory, "Event handler factory");
         final int workerCount = ioReactorConfig != null ? ioReactorConfig.getIoThreadCount() : IOReactorConfig.DEFAULT.getIoThreadCount();
         this.workers = new SingleCoreIOReactor[workerCount];
@@ -81,7 +103,8 @@ public DefaultConnectingIOReactor(
                     ioSessionDecorator,
                     sessionListener,
                     threadPoolListener,
-                    sessionShutdownCallback);
+                    sessionShutdownCallback,
+                    socketChannelFactory);
             this.workers[i] = dispatcher;
             threads[i] = (threadFactory != null ? threadFactory : THREAD_FACTORY).newThread(new IOReactorWorker(dispatcher));
         }
@@ -101,13 +124,43 @@ public DefaultConnectingIOReactor(
                 null, sessionShutdownCallback, null);
     }
 
+    /**
+     * @since 5.5
+     */
+    public DefaultConnectingIOReactor(
+            final IOEventHandlerFactory eventHandlerFactory,
+            final IOReactorConfig ioReactorConfig,
+            final ThreadFactory threadFactory,
+            final Decorator<IOSession> ioSessionDecorator,
+            final Callback<Exception> exceptionCallback,
+            final IOSessionListener sessionListener,
+            final Callback<IOSession> sessionShutdownCallback,
+            final IOFunction<SocketAddress, SocketChannel> socketChannelFactory) {
+        this(eventHandlerFactory, ioReactorConfig, threadFactory, ioSessionDecorator, exceptionCallback, sessionListener,
+                null, sessionShutdownCallback, null, socketChannelFactory);
+    }
+
     public DefaultConnectingIOReactor(
             final IOEventHandlerFactory eventHandlerFactory,
             final IOReactorConfig config,
             final Callback<IOSession> sessionShutdownCallback) {
         this(eventHandlerFactory, config, null, null, null, null, sessionShutdownCallback);
     }
 
+    /**
+     * Creates an instance with a custom socket channel factory.
+     *
+     * @since 5.5
+     */
+    public DefaultConnectingIOReactor(
+            final IOEventHandlerFactory eventHandlerFactory,
+            final IOReactorConfig config,
+            final Callback<IOSession> sessionShutdownCallback,
+            final IOFunction<SocketAddress, SocketChannel> socketChannelFactory) {
+        this(eventHandlerFactory, config, null, null, null, null, null, sessionShutdownCallback, null,
+                socketChannelFactory);
+    }
+
     /**
      * Creates an instance of DefaultConnectingIOReactor with default configuration.
      *

httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java

@@ -54,6 +54,7 @@
 import org.apache.hc.core5.function.Decorator;
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.hc.core5.io.Closer;
+import org.apache.hc.core5.io.IOFunction;
 import org.apache.hc.core5.net.NamedEndpoint;
 import org.apache.hc.core5.util.Args;
 import org.apache.hc.core5.util.ReflectionUtils;
@@ -77,6 +78,7 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect
     private volatile long lastTimeoutCheckNanos;
     private volatile long lastSelectNanos;
     private final IOReactorMetricsListener threadPoolListener;
+    private final IOFunction<SocketAddress, SocketChannel> socketChannelFactory;
 
     // Atomic variables for tracking total wait time and count of processed requests
     private final AtomicLong totalWaitTime = new AtomicLong(0);
@@ -90,13 +92,27 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect
             final IOSessionListener sessionListener,
             final IOReactorMetricsListener threadPoolListener,
             final Callback<IOSession> sessionShutdownCallback) {
+        this(exceptionCallback, eventHandlerFactory, reactorConfig, ioSessionDecorator, sessionListener,
+                threadPoolListener, sessionShutdownCallback, null);
+    }
+
+    SingleCoreIOReactor(
+            final Callback<Exception> exceptionCallback,
+            final IOEventHandlerFactory eventHandlerFactory,
+            final IOReactorConfig reactorConfig,
+            final Decorator<IOSession> ioSessionDecorator,
+            final IOSessionListener sessionListener,
+            final IOReactorMetricsListener threadPoolListener,
+            final Callback<IOSession> sessionShutdownCallback,
+            final IOFunction<SocketAddress, SocketChannel> socketChannelFactory) {
         super(exceptionCallback);
         this.eventHandlerFactory = Args.notNull(eventHandlerFactory, "Event handler factory");
         this.reactorConfig = Args.notNull(reactorConfig, "I/O reactor config");
         this.ioSessionDecorator = ioSessionDecorator;
         this.sessionListener = sessionListener;
         this.threadPoolListener = threadPoolListener;
         this.sessionShutdownCallback = sessionShutdownCallback;
+        this.socketChannelFactory = socketChannelFactory != null ? socketChannelFactory : SingleCoreIOReactor::openSocketFor;
         this.shutdownInitiated = new AtomicBoolean();
         this.closedSessions = new ConcurrentLinkedQueue<>();
         this.channelQueue = new ConcurrentLinkedQueue<>();
@@ -349,8 +365,8 @@ private void processPendingConnectionRequests() {
             if (!sessionRequest.isCancelled()) {
                 final SocketChannel socketChannel;
                 try {
-                    socketChannel = openSocketFor(sessionRequest.remoteAddress);
-                } catch (final IOException ex) {
+                    socketChannel = socketChannelFactory.apply(sessionRequest.remoteAddress);
+                } catch (final IOException | RuntimeException ex) {
                     sessionRequest.failed(ex);
                     return;
                 }

httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSocketChannelFactory.java

@@ -0,0 +1,142 @@
+/*
+ * ====================================================================
+ * 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.core5.reactor;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hc.core5.io.IOFunction;
+import org.apache.hc.core5.net.NamedEndpoint;
+import org.apache.hc.core5.util.Timeout;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class TestSocketChannelFactory {
+
+    @Test
+    void testCustomSocketChannelFactory() throws Exception {
+        final AtomicReference<SocketAddress> requestedAddress = new AtomicReference<>();
+        try (ServerSocketChannel serverChannel = ServerSocketChannel.open()) {
+            serverChannel.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
+            final InetSocketAddress remoteAddress = (InetSocketAddress) serverChannel.getLocalAddress();
+            final IOFunction<SocketAddress, SocketChannel> socketChannelFactory = address -> {
+                requestedAddress.set(address);
+                return SocketChannel.open();
+            };
+            final IOEventHandler ioEventHandler = Mockito.mock(IOEventHandler.class);
+            final IOEventHandlerFactory eventHandlerFactory = (ioSession, attachment) -> ioEventHandler;
+            final IOReactorConfig config = IOReactorConfig.custom().setIoThreadCount(1).build();
+            final DefaultConnectingIOReactor reactor = new DefaultConnectingIOReactor(
+                    eventHandlerFactory, config, null, socketChannelFactory);
+            try {
+                reactor.start();
+                final Future<IOSession> future = reactor.connect(
+                        endpoint(remoteAddress),
+                        remoteAddress,
+                        null,
+                        Timeout.ofSeconds(1),
+                        null,
+                        null);
+                final IOSession ioSession = future.get(5, TimeUnit.SECONDS);
+                Assertions.assertNotNull(ioSession);
+                Assertions.assertEquals(remoteAddress, requestedAddress.get());
+            } finally {
+                reactor.close();
+            }
+        }
+    }
+
+    @Test
+    void testIOExceptionFromCustomSocketChannelFactory() throws Exception {
+        final IOException expected = new IOException("custom transport");
+        final IOFunction<SocketAddress, SocketChannel> socketChannelFactory = remoteAddress -> {
+            throw expected;
+        };
+        assertFactoryFailure(socketChannelFactory, expected);
+    }
+
+    @Test
+    void testRuntimeExceptionFromCustomSocketChannelFactory() throws Exception {
+        final RuntimeException expected = new IllegalStateException("custom transport");
+        final IOFunction<SocketAddress, SocketChannel> socketChannelFactory = remoteAddress -> {
+            throw expected;
+        };
+        assertFactoryFailure(socketChannelFactory, expected);
+    }
+
+    private static void assertFactoryFailure(
+            final IOFunction<SocketAddress, SocketChannel> socketChannelFactory,
+            final Exception expected) throws Exception {
+        final IOEventHandlerFactory eventHandlerFactory = Mockito.mock(IOEventHandlerFactory.class);
+        final IOReactorConfig config = IOReactorConfig.custom().setIoThreadCount(1).build();
+        final DefaultConnectingIOReactor reactor = new DefaultConnectingIOReactor(
+                eventHandlerFactory, config, null, socketChannelFactory);
+        try {
+            reactor.start();
+            final InetSocketAddress remoteAddress = new InetSocketAddress(InetAddress.getLoopbackAddress(), 80);
+            final Future<IOSession> future = reactor.connect(
+                    endpoint(remoteAddress),
+                    remoteAddress,
+                    null,
+                    Timeout.ofSeconds(1),
+                    null,
+                    null);
+            final ExecutionException ex = Assertions.assertThrows(
+                    ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
+            Assertions.assertSame(expected, ex.getCause());
+        } finally {
+            reactor.close();
+        }
+    }
+
+    private static NamedEndpoint endpoint(final InetSocketAddress address) {
+        return new NamedEndpoint() {
+
+            @Override
+            public String getHostName() {
+                return address.getHostString();
+            }
+
+            @Override
+            public int getPort() {
+                return address.getPort();
+            }
+
+        };
+    }
+
+}