diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java index fb1ae47c8..5389eebe0 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java @@ -28,6 +28,8 @@ 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; @@ -35,6 +37,7 @@ 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 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 ioSessionDecorator, + final Callback exceptionCallback, + final IOSessionListener sessionListener, + final IOReactorMetricsListener threadPoolListener, + final Callback sessionShutdownCallback, + final IOWorkerSelector workerSelector, + final IOFunction 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,6 +124,22 @@ public DefaultConnectingIOReactor( null, sessionShutdownCallback, null); } + /** + * @since 5.5 + */ + public DefaultConnectingIOReactor( + final IOEventHandlerFactory eventHandlerFactory, + final IOReactorConfig ioReactorConfig, + final ThreadFactory threadFactory, + final Decorator ioSessionDecorator, + final Callback exceptionCallback, + final IOSessionListener sessionListener, + final Callback sessionShutdownCallback, + final IOFunction socketChannelFactory) { + this(eventHandlerFactory, ioReactorConfig, threadFactory, ioSessionDecorator, exceptionCallback, sessionListener, + null, sessionShutdownCallback, null, socketChannelFactory); + } + public DefaultConnectingIOReactor( final IOEventHandlerFactory eventHandlerFactory, final IOReactorConfig config, @@ -108,6 +147,20 @@ public DefaultConnectingIOReactor( 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 sessionShutdownCallback, + final IOFunction socketChannelFactory) { + this(eventHandlerFactory, config, null, null, null, null, null, sessionShutdownCallback, null, + socketChannelFactory); + } + /** * Creates an instance of DefaultConnectingIOReactor with default configuration. * diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java index cd2b278a6..e5839b3e8 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java +++ b/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 socketChannelFactory; // Atomic variables for tracking total wait time and count of processed requests private final AtomicLong totalWaitTime = new AtomicLong(0); @@ -90,6 +92,19 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect final IOSessionListener sessionListener, final IOReactorMetricsListener threadPoolListener, final Callback sessionShutdownCallback) { + this(exceptionCallback, eventHandlerFactory, reactorConfig, ioSessionDecorator, sessionListener, + threadPoolListener, sessionShutdownCallback, null); + } + + SingleCoreIOReactor( + final Callback exceptionCallback, + final IOEventHandlerFactory eventHandlerFactory, + final IOReactorConfig reactorConfig, + final Decorator ioSessionDecorator, + final IOSessionListener sessionListener, + final IOReactorMetricsListener threadPoolListener, + final Callback sessionShutdownCallback, + final IOFunction socketChannelFactory) { super(exceptionCallback); this.eventHandlerFactory = Args.notNull(eventHandlerFactory, "Event handler factory"); this.reactorConfig = Args.notNull(reactorConfig, "I/O reactor config"); @@ -97,6 +112,7 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect 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; } diff --git a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSocketChannelFactory.java b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSocketChannelFactory.java new file mode 100644 index 000000000..2036a2e16 --- /dev/null +++ b/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 + * . + * + */ + +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 requestedAddress = new AtomicReference<>(); + try (ServerSocketChannel serverChannel = ServerSocketChannel.open()) { + serverChannel.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)); + final InetSocketAddress remoteAddress = (InetSocketAddress) serverChannel.getLocalAddress(); + final IOFunction 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 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 socketChannelFactory = remoteAddress -> { + throw expected; + }; + assertFactoryFailure(socketChannelFactory, expected); + } + + @Test + void testRuntimeExceptionFromCustomSocketChannelFactory() throws Exception { + final RuntimeException expected = new IllegalStateException("custom transport"); + final IOFunction socketChannelFactory = remoteAddress -> { + throw expected; + }; + assertFactoryFailure(socketChannelFactory, expected); + } + + private static void assertFactoryFailure( + final IOFunction 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 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(); + } + + }; + } + +}