Expose ioThreadsCount config property (AsyncHttpClient#1283)

Author: Spikhalskiy

client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -288,6 +288,8 @@ public interface AsyncHttpClientConfig {
 
     ByteBufAllocator getAllocator();
 
+    int getIoThreadsCount();
+
     interface AdditionalChannelInitializer {
 
         void initChannel(Channel channel) throws Exception;

client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java

@@ -131,6 +131,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig {
     private final AdditionalChannelInitializer httpAdditionalChannelInitializer;
     private final AdditionalChannelInitializer wsAdditionalChannelInitializer;
     private final ResponseBodyPartFactory responseBodyPartFactory;
+    private final int ioThreadsCount;
 
     private DefaultAsyncHttpClientConfig(//
             // http
@@ -203,7 +204,8 @@ private DefaultAsyncHttpClientConfig(//
             ThreadFactory threadFactory,//
             AdditionalChannelInitializer httpAdditionalChannelInitializer,//
             AdditionalChannelInitializer wsAdditionalChannelInitializer,//
-            ResponseBodyPartFactory responseBodyPartFactory) {
+            ResponseBodyPartFactory responseBodyPartFactory,//
+            int ioThreadsCount) {
 
         // http
         this.followRedirect = followRedirect;
@@ -276,6 +278,7 @@ private DefaultAsyncHttpClientConfig(//
         this.httpAdditionalChannelInitializer = httpAdditionalChannelInitializer;
         this.wsAdditionalChannelInitializer = wsAdditionalChannelInitializer;
         this.responseBodyPartFactory = responseBodyPartFactory;
+        this.ioThreadsCount = ioThreadsCount;
     }
 
     @Override
@@ -581,6 +584,11 @@ public ResponseBodyPartFactory getResponseBodyPartFactory() {
         return responseBodyPartFactory;
     }
 
+    @Override
+    public int getIoThreadsCount() {
+        return ioThreadsCount;
+    }
+
     /**
      * Builder for an {@link AsyncHttpClient}
      */
@@ -659,6 +667,7 @@ public static class Builder {
         private AdditionalChannelInitializer httpAdditionalChannelInitializer;
         private AdditionalChannelInitializer wsAdditionalChannelInitializer;
         private ResponseBodyPartFactory responseBodyPartFactory = ResponseBodyPartFactory.EAGER;
+        private int ioThreadsCount = defaultIoThreadsCount();
 
         public Builder() {
         }
@@ -732,6 +741,7 @@ public Builder(AsyncHttpClientConfig config) {
             httpAdditionalChannelInitializer = config.getHttpAdditionalChannelInitializer();
             wsAdditionalChannelInitializer = config.getWsAdditionalChannelInitializer();
             responseBodyPartFactory = config.getResponseBodyPartFactory();
+            ioThreadsCount = config.getIoThreadsCount();
         }
 
         // http
@@ -1066,6 +1076,11 @@ public Builder setResponseBodyPartFactory(ResponseBodyPartFactory responseBodyPa
             return this;
         }
 
+        public Builder setIoThreadsCount(int ioThreadsCount) {
+            this.ioThreadsCount = ioThreadsCount;
+            return this;
+        }
+
         private ProxyServerSelector resolveProxyServerSelector() {
             if (proxyServerSelector != null)
                 return proxyServerSelector;
@@ -1139,7 +1154,8 @@ public DefaultAsyncHttpClientConfig build() {
                     threadFactory, //
                     httpAdditionalChannelInitializer, //
                     wsAdditionalChannelInitializer, //
-                    responseBodyPartFactory);
+                    responseBodyPartFactory, //
+                    ioThreadsCount);
         }
     }
 }

client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java

@@ -197,4 +197,8 @@ public static int defaultShutdownTimeout() {
     public static boolean defaultUseNativeTransport() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "useNativeTransport");
     }
+
+    public static int defaultIoThreadsCount() {
+        return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getInt(ASYNC_CLIENT_CONFIG_ROOT + "ioThreadsCount");
+    }
 }

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -166,11 +166,11 @@ public boolean remove(Object o) {
         ChannelFactory<? extends Channel> channelFactory;
         if (allowReleaseEventLoopGroup) {
             if (config.isUseNativeTransport()) {
-                eventLoopGroup = newEpollEventLoopGroup(threadFactory);
+                eventLoopGroup = newEpollEventLoopGroup(config.getIoThreadsCount(), threadFactory);
                 channelFactory = getEpollSocketChannelFactory();
 
             } else {
-                eventLoopGroup = new NioEventLoopGroup(0, threadFactory);
+                eventLoopGroup = new NioEventLoopGroup(config.getIoThreadsCount(), threadFactory);
                 channelFactory = NioSocketChannelFactory.INSTANCE;
             }
 
@@ -224,10 +224,10 @@ private Bootstrap newBootstrap(ChannelFactory<? extends Channel> channelFactory,
         return bootstrap;
     }
 
-    private EventLoopGroup newEpollEventLoopGroup(ThreadFactory threadFactory) {
+    private EventLoopGroup newEpollEventLoopGroup(int ioThreadsCount, ThreadFactory threadFactory) {
         try {
             Class<?> epollEventLoopGroupClass = Class.forName("io.netty.channel.epoll.EpollEventLoopGroup");
-            return (EventLoopGroup) epollEventLoopGroupClass.getConstructor(int.class, ThreadFactory.class).newInstance(0, threadFactory);
+            return (EventLoopGroup) epollEventLoopGroupClass.getConstructor(int.class, ThreadFactory.class).newInstance(ioThreadsCount, threadFactory);
         } catch (Exception e) {
             throw new IllegalArgumentException(e);
         }

client/src/main/resources/ahc-default.properties

@@ -42,3 +42,4 @@ org.asynchttpclient.keepEncodingHeader=false
 org.asynchttpclient.shutdownQuietPeriod=2000
 org.asynchttpclient.shutdownTimeout=15000
 org.asynchttpclient.useNativeTransport=false
+org.asynchttpclient.ioThreadsCount=0