From a49ac074492164f1f4669f8ec538cd6ed57f0a22 Mon Sep 17 00:00:00 2001 From: Aayush Atharva <24762260+hyperxpro@users.noreply.github.com> Date: Sun, 20 Sep 2026 19:25:23 +0000 Subject: [PATCH] Detect transport of external event loop group by IoHandler --- .../netty/channel/ChannelManager.java | 53 +++++++++------ .../DefaultAsyncHttpClientTest.java | 64 +++++++++++++++++++ 2 files changed, 98 insertions(+), 19 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index 9a2f1a19c..e571360e6 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -27,10 +27,11 @@ import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; import io.netty.channel.EventLoopGroup; +import io.netty.channel.IoEventLoopGroup; +import io.netty.channel.IoHandler; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.ChannelGroupFuture; import io.netty.channel.group.DefaultChannelGroup; -import io.netty.channel.nio.NioEventLoopGroup; import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.websocketx.WebSocket08FrameDecoder; @@ -172,14 +173,15 @@ public class ChannelManager { private AsyncHttpClientHandler wsHandler; private Http2Handler http2Handler; - private boolean isInstanceof(Object object, String name) { + // Handler classes are looked up by name, and not initialized, because the native transports are optional deps. + private boolean isIoType(IoEventLoopGroup eventLoopGroup, String ioHandlerClassName) { final Class clazz; try { - clazz = Class.forName(name, false, getClass().getClassLoader()); + clazz = Class.forName(ioHandlerClassName, false, getClass().getClassLoader()); } catch (ClassNotFoundException ignored) { return false; } - return clazz.isInstance(object); + return IoHandler.class.isAssignableFrom(clazz) && eventLoopGroup.isIoType(clazz.asSubclass(IoHandler.class)); } public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { @@ -207,11 +209,12 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { // check if external EventLoopGroup is defined ThreadFactory threadFactory = config.getThreadFactory() != null ? config.getThreadFactory() : new DefaultThreadFactory(config.getThreadPoolName()); - allowReleaseEventLoopGroup = config.getEventLoopGroup() == null; + EventLoopGroup configuredEventLoopGroup = config.getEventLoopGroup(); + allowReleaseEventLoopGroup = configuredEventLoopGroup == null; TransportFactory transportFactory; EventLoopGroup localEventLoopGroup; - if (allowReleaseEventLoopGroup) { + if (configuredEventLoopGroup == null) { if (config.isUseNativeTransport()) { transportFactory = getNativeTransportFactory(config); } else { @@ -234,19 +237,8 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { } } } else { - localEventLoopGroup = config.getEventLoopGroup(); - - if (localEventLoopGroup instanceof NioEventLoopGroup) { - transportFactory = NioTransportFactory.INSTANCE; - } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.epoll.EpollEventLoopGroup")) { - transportFactory = new EpollTransportFactory(); - } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.kqueue.KQueueEventLoopGroup")) { - transportFactory = new KQueueTransportFactory(); - } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.uring.IOUringEventLoopGroup")) { - transportFactory = new IoUringTransportFactory(); - } else { - throw new IllegalArgumentException("Unknown event loop group " + localEventLoopGroup.getClass().getSimpleName()); - } + localEventLoopGroup = configuredEventLoopGroup; + transportFactory = transportFactoryOf(configuredEventLoopGroup); } this.eventLoopGroup = localEventLoopGroup; @@ -258,6 +250,29 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { addressResolverGroup = config.getAddressResolverGroup(); } + // Since Netty 4.2 a group is a MultiThreadIoEventLoopGroup parameterized with an IoHandler, so its class no + // longer tells the transport apart: the per-transport group types are deprecated and io_uring never had one + // (issue #2121). Ask the group which handler it runs, as our channels must be registrable on it. + private TransportFactory transportFactoryOf(EventLoopGroup eventLoopGroup) { + if (eventLoopGroup instanceof IoEventLoopGroup) { + IoEventLoopGroup ioEventLoopGroup = (IoEventLoopGroup) eventLoopGroup; + if (isIoType(ioEventLoopGroup, "io.netty.channel.nio.NioIoHandler")) { + return NioTransportFactory.INSTANCE; + } + if (isIoType(ioEventLoopGroup, "io.netty.channel.epoll.EpollIoHandler")) { + return new EpollTransportFactory(); + } + if (isIoType(ioEventLoopGroup, "io.netty.channel.kqueue.KQueueIoHandler")) { + return new KQueueTransportFactory(); + } + if (isIoType(ioEventLoopGroup, "io.netty.channel.uring.IoUringIoHandler")) { + return new IoUringTransportFactory(); + } + } + throw new IllegalArgumentException("Unsupported event loop group " + eventLoopGroup.getClass().getName() + + ": it runs no known IoHandler (NIO, epoll, kqueue or io_uring)"); + } + private static TransportFactory getNativeTransportFactory(AsyncHttpClientConfig config) { // If we are running on macOS then use KQueue if (PlatformDependent.isOsx()) { diff --git a/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java b/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java index 0070e06b4..2e7ea386b 100644 --- a/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java +++ b/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java @@ -16,13 +16,19 @@ package org.asynchttpclient; import io.github.artsok.RepeatedIfExceptionsTest; +import io.netty.channel.DefaultEventLoopGroup; import io.netty.channel.EventLoopGroup; import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.epoll.Epoll; import io.netty.channel.epoll.EpollEventLoopGroup; +import io.netty.channel.epoll.EpollIoHandler; +import io.netty.channel.kqueue.KQueue; import io.netty.channel.kqueue.KQueueEventLoopGroup; +import io.netty.channel.kqueue.KQueueIoHandler; import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.uring.IoUring; +import io.netty.channel.uring.IoUringIoHandler; import io.netty.util.Timer; import org.asynchttpclient.cookie.CookieEvictionTask; import org.asynchttpclient.cookie.CookieStore; @@ -41,7 +47,9 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assumptions.assumeTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; @@ -75,6 +83,62 @@ public void testNativeTransportKQueueOnMacOs() throws Exception { assertRequestSucceedsAndEventLoopGroupIs(config, KQueueEventLoopGroup.class); } + @RepeatedIfExceptionsTest(repeats = 5) + public void testExternalNioEventLoopGroup() throws Exception { + assertRequestSucceedsWith(new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory())); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void testExternalDeprecatedNioEventLoopGroup() throws Exception { + assertRequestSucceedsWith(new NioEventLoopGroup(1)); + } + + @RepeatedIfExceptionsTest(repeats = 5) + @EnabledOnOs(OS.LINUX) + public void testExternalEpollEventLoopGroup() throws Exception { + assumeTrue(Epoll.isAvailable(), "epoll is not available"); + assertRequestSucceedsWith(new MultiThreadIoEventLoopGroup(1, EpollIoHandler.newFactory())); + } + + @RepeatedIfExceptionsTest(repeats = 5) + @EnabledOnOs(OS.LINUX) + public void testExternalIoUringEventLoopGroup() throws Exception { + assumeTrue(IoUring.isAvailable(), "io_uring is not available"); + assertRequestSucceedsWith(new MultiThreadIoEventLoopGroup(1, IoUringIoHandler.newFactory())); + } + + @RepeatedIfExceptionsTest(repeats = 5) + @EnabledOnOs(OS.MAC) + public void testExternalKQueueEventLoopGroup() throws Exception { + assumeTrue(KQueue.isAvailable(), "kqueue is not available"); + assertRequestSucceedsWith(new MultiThreadIoEventLoopGroup(1, KQueueIoHandler.newFactory())); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void testExternalEventLoopGroupOfUnknownTransportIsRejected() throws Exception { + EventLoopGroup eventLoopGroup = new DefaultEventLoopGroup(1); + try { + AsyncHttpClientConfig config = config().setEventLoopGroup(eventLoopGroup).build(); + assertThrows(IllegalArgumentException.class, () -> asyncHttpClient(config)); + } finally { + eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS).await(); + } + } + + private static void assertRequestSucceedsWith(EventLoopGroup eventLoopGroup) throws Exception { + try (HttpServer server = new HttpServer()) { + server.start(); + server.enqueueOk(); + AsyncHttpClientConfig config = config().setEventLoopGroup(eventLoopGroup).build(); + try (DefaultAsyncHttpClient client = (DefaultAsyncHttpClient) asyncHttpClient(config)) { + assertDoesNotThrow(() -> client.prepareGet(server.getHttpUrl()).execute().get()); + assertSame(eventLoopGroup, client.channelManager().getEventLoopGroup()); + } + } finally { + eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS).await(); + } + } + private static void assertRequestSucceedsAndEventLoopGroupIs(AsyncHttpClientConfig config, Class expectedEventLoopGroupType) throws Exception { try (HttpServer server = new HttpServer(); DefaultAsyncHttpClient client = (DefaultAsyncHttpClient) asyncHttpClient(config)) {