Add a config prop to control memory pooling

Author: slandelle

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

@@ -257,6 +257,8 @@ public interface AsyncHttpClientConfig {
 
     boolean isValidateResponseHeaders();
 
+    boolean isUsePooledMemory();
+
     interface AdditionalChannelInitializer {
 
         void initChannel(Channel channel) throws Exception;

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

@@ -117,6 +117,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig {
     private final Map<ChannelOption<Object>, Object> channelOptions;
     private final EventLoopGroup eventLoopGroup;
     private final boolean useNativeTransport;
+    private final boolean usePooledMemory;
     private final Timer nettyTimer;
     private final ThreadFactory threadFactory;
     private final AdditionalChannelInitializer httpAdditionalChannelInitializer;
@@ -181,6 +182,7 @@ private DefaultAsyncHttpClientConfig(//
             Map<ChannelOption<Object>, Object> channelOptions,//
             EventLoopGroup eventLoopGroup,//
             boolean useNativeTransport,//
+            boolean usePooledMemory,//
             Timer nettyTimer,//
             ThreadFactory threadFactory,//
             AdditionalChannelInitializer httpAdditionalChannelInitializer,//
@@ -244,6 +246,7 @@ private DefaultAsyncHttpClientConfig(//
         this.channelOptions = channelOptions;
         this.eventLoopGroup = eventLoopGroup;
         this.useNativeTransport = useNativeTransport;
+        this.usePooledMemory = usePooledMemory;
         this.nettyTimer = nettyTimer;
         this.threadFactory = threadFactory;
         this.httpAdditionalChannelInitializer = httpAdditionalChannelInitializer;
@@ -493,6 +496,11 @@ public boolean isUseNativeTransport() {
         return useNativeTransport;
     }
 
+    @Override
+    public boolean isUsePooledMemory() {
+        return usePooledMemory;
+    }
+
     @Override
     public Timer getNettyTimer() {
         return nettyTimer;
@@ -580,6 +588,7 @@ public static class Builder {
         private int webSocketMaxBufferSize = defaultWebSocketMaxBufferSize();
         private int webSocketMaxFrameSize = defaultWebSocketMaxFrameSize();
         private boolean useNativeTransport = defaultUseNativeTransport();
+        private boolean usePooledMemory = defaultUsePooledMemory();
         private Map<ChannelOption<Object>, Object> channelOptions = new HashMap<>();
         private EventLoopGroup eventLoopGroup;
         private Timer nettyTimer;
@@ -647,6 +656,7 @@ public Builder(AsyncHttpClientConfig config) {
             channelOptions.putAll(config.getChannelOptions());
             eventLoopGroup = config.getEventLoopGroup();
             useNativeTransport = config.isUseNativeTransport();
+            usePooledMemory = config.isUsePooledMemory();
             nettyTimer = config.getNettyTimer();
             threadFactory = config.getThreadFactory();
             httpAdditionalChannelInitializer = config.getHttpAdditionalChannelInitializer();
@@ -1018,6 +1028,7 @@ public DefaultAsyncHttpClientConfig build() {
                     channelOptions.isEmpty() ? Collections.emptyMap() : Collections.unmodifiableMap(channelOptions),//
                     eventLoopGroup, //
                     useNativeTransport, //
+                    usePooledMemory, //
                     nettyTimer, //
                     threadFactory, //
                     httpAdditionalChannelInitializer, //

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

@@ -78,7 +78,7 @@ public static boolean defaultUseProxySelector() {
     public static boolean defaultUseProxyProperties() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "useProxyProperties");
     }
-    
+
     public static boolean defaultValidateResponseHeaders() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "validateResponseHeaders");
     }
@@ -162,4 +162,8 @@ public static int defaultShutdownTimeout() {
     public static boolean defaultUseNativeTransport() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "useNativeTransport");
     }
+
+    public static boolean defaultUsePooledMemory() {
+        return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "usePooledMemory");
+    }
 }

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

@@ -16,6 +16,7 @@
 import static org.asynchttpclient.util.MiscUtils.trimStackTrace;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
@@ -192,7 +193,7 @@ private Bootstrap newBootstrap(Class<? extends Channel> socketChannelClass, Even
         @SuppressWarnings("deprecation")
         Bootstrap bootstrap = new Bootstrap().channel(socketChannelClass).group(eventLoopGroup)//
                 // default to PooledByteBufAllocator
-                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)//
+                .option(ChannelOption.ALLOCATOR, config.isUsePooledMemory() ? PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT)//
                 .option(ChannelOption.TCP_NODELAY, true)//
                 .option(ChannelOption.AUTO_CLOSE, false);
 

client/src/main/java/org/asynchttpclient/netty/request/body/BodyChunkedInput.java

@@ -15,7 +15,6 @@
 
 import static org.asynchttpclient.util.Assertions.assertNotNull;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.stream.ChunkedInput;
 
@@ -48,19 +47,20 @@ public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
         if (endOfInput)
             return null;
 
-        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(chunkSize);
+        ByteBuf buffer = ctx.alloc().buffer(chunkSize);
+
         Body.BodyState state = body.transferTo(buffer);
         switch (state) {
-            case STOP:
-                endOfInput = true;
-                return buffer;
-            case SUSPEND:
-                //this will suspend the stream in ChunkedWriteHandler
-                return null;
-            case CONTINUE:
-                return buffer;
-            default:
-                throw new IllegalStateException("Unknown state: " + state);
+        case STOP:
+            endOfInput = true;
+            return buffer;
+        case SUSPEND:
+            // this will suspend the stream in ChunkedWriteHandler
+            return null;
+        case CONTINUE:
+            return buffer;
+        default:
+            throw new IllegalStateException("Unknown state: " + state);
         }
     }
 

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

@@ -36,3 +36,4 @@ org.asynchttpclient.keepEncodingHeader=false
 org.asynchttpclient.shutdownQuietPeriod=2000
 org.asynchttpclient.shutdownTimeout=15000
 org.asynchttpclient.useNativeTransport=false
+org.asynchttpclient.usePooledMemory=true