Extract HttpResponseBodyPart creation into a factory so one can chose to bypass auto bytes extraction

Author: Stephane Landelle

api/src/main/java/org/asynchttpclient/HttpResponseBodyPart.java

@@ -75,13 +75,13 @@ public abstract class HttpResponseBodyPart {
      * underlying TCP connection will be closed as soon as the processing of the response is completed. That
      * means the underlying connection will never get pooled.
      */
-    public abstract void markUnderlyingConnectionAsClosed();
+    public abstract void markUnderlyingConnectionAsToBeClosed();
 
     /**
      * Return true of the underlying connection will be closed once the response has been fully processed.
      *
      * @return true of the underlying connection will be closed once the response has been fully processed.
      */
-    public abstract boolean closeUnderlyingConnection();
+    public abstract boolean isUnderlyingConnectionToBeClosed();
 
 }

api/src/test/java/org/asynchttpclient/async/AsyncStreamHandlerTest.java

@@ -546,7 +546,7 @@ public STATE onBodyPartReceived(HttpResponseBodyPart content) throws Exception {
                     builder.accumulate(content);
 
                     if (content.isLast()) {
-                        content.markUnderlyingConnectionAsClosed();
+                        content.markUnderlyingConnectionAsToBeClosed();
                     }
                     return STATE.CONTINUE;
                 }

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/GrizzlyResponseBodyPart.java

@@ -122,15 +122,15 @@ public boolean isLast() {
      * {@inheritDoc}
      */
     @Override
-    public void markUnderlyingConnectionAsClosed() {
+    public void markUnderlyingConnectionAsToBeClosed() {
         ConnectionManager.markConnectionAsDoNotCache(connection);
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
-    public boolean closeUnderlyingConnection() {
+    public boolean isUnderlyingConnectionToBeClosed() {
         return !ConnectionManager.isConnectionCacheable(connection);
     }
 

providers/netty/src/main/java/org/asynchttpclient/providers/netty/NettyAsyncHttpProvider.java

@@ -34,7 +34,7 @@ public class NettyAsyncHttpProvider implements AsyncHttpProvider {
     private static final Logger LOGGER = LoggerFactory.getLogger(NettyAsyncHttpProvider.class);
 
     private final AsyncHttpClientConfig config;
-    private final NettyAsyncHttpProviderConfig asyncHttpProviderConfig;
+    private final NettyAsyncHttpProviderConfig nettyConfig;
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final Channels channels;
     private final NettyRequestSender requestSender;
@@ -43,13 +43,13 @@ public class NettyAsyncHttpProvider implements AsyncHttpProvider {
     public NettyAsyncHttpProvider(AsyncHttpClientConfig config) {
 
         this.config = config;
-        asyncHttpProviderConfig = config.getAsyncHttpProviderConfig() instanceof NettyAsyncHttpProviderConfig ? //
+        nettyConfig = config.getAsyncHttpProviderConfig() instanceof NettyAsyncHttpProviderConfig ? //
         NettyAsyncHttpProviderConfig.class.cast(config.getAsyncHttpProviderConfig())
                 : new NettyAsyncHttpProviderConfig();
 
-        channels = new Channels(config, asyncHttpProviderConfig);
+        channels = new Channels(config, nettyConfig);
         requestSender = new NettyRequestSender(closed, config, channels);
-        channelHandler = new NettyChannelHandler(config, requestSender, channels, closed);
+        channelHandler = new NettyChannelHandler(config, nettyConfig, requestSender, channels, closed);
         channels.configure(channelHandler);
     }
 
@@ -72,6 +72,6 @@ public void close() {
 
     @Override
     public <T> ListenableFuture<T> execute(Request request, final AsyncHandler<T> asyncHandler) throws IOException {
-        return requestSender.sendRequest(request, asyncHandler, null, asyncHttpProviderConfig.isAsyncConnect(), false);
+        return requestSender.sendRequest(request, asyncHandler, null, nettyConfig.isAsyncConnect(), false);
     }
 }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/NettyAsyncHttpProviderConfig.java

@@ -16,6 +16,7 @@
  */
 package org.asynchttpclient.providers.netty;
 
+import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
@@ -25,6 +26,10 @@
 import java.util.Set;
 
 import org.asynchttpclient.AsyncHttpProviderConfig;
+import org.asynchttpclient.providers.netty.response.DefaultResponseBodyPart;
+import org.asynchttpclient.providers.netty.response.LazyResponseBodyPart;
+import org.asynchttpclient.providers.netty.response.ResponseBodyPart;
+import org.asynchttpclient.providers.netty.util.ByteBufUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,7 +49,7 @@ public class NettyAsyncHttpProviderConfig implements AsyncHttpProviderConfig<Str
      * Allow configuring the Netty's event loop.
      */
     private EventLoopGroup eventLoopGroup;
-    
+
     private AdditionalChannelInitializer httpAdditionalChannelInitializer;
     private AdditionalChannelInitializer wsAdditionalChannelInitializer;
     private AdditionalChannelInitializer httpsAdditionalChannelInitializer;
@@ -87,6 +92,8 @@ public class NettyAsyncHttpProviderConfig implements AsyncHttpProviderConfig<Str
 
     private final Map<String, Object> properties = new HashMap<String, Object>();
 
+    private ResponseBodyPartFactory bodyPartFactory = new DefaultResponseBodyPartFactory();
+
     public NettyAsyncHttpProviderConfig() {
         properties.put(REUSE_ADDRESS, Boolean.FALSE);
     }
@@ -219,9 +226,38 @@ public AdditionalChannelInitializer getWssAdditionalChannelInitializer() {
     public void setWssAdditionalChannelInitializer(AdditionalChannelInitializer wssAdditionalChannelInitializer) {
         this.wssAdditionalChannelInitializer = wssAdditionalChannelInitializer;
     }
-    
+
+    public ResponseBodyPartFactory getBodyPartFactory() {
+        return bodyPartFactory;
+    }
+
+    public void setBodyPartFactory(ResponseBodyPartFactory bodyPartFactory) {
+        this.bodyPartFactory = bodyPartFactory;
+    }
+
     public static interface AdditionalChannelInitializer {
 
         void initChannel(Channel ch) throws Exception;
     }
+
+    public static interface ResponseBodyPartFactory {
+
+        ResponseBodyPart newResponseBodyPart(ByteBuf buf, boolean last);
+    }
+
+    public static class DefaultResponseBodyPartFactory implements ResponseBodyPartFactory {
+
+        @Override
+        public ResponseBodyPart newResponseBodyPart(ByteBuf buf, boolean last) {
+            return new DefaultResponseBodyPart(ByteBufUtil.byteBuf2Bytes(buf), last);
+        }
+    }
+
+    public static class LazyResponseBodyPartFactory implements ResponseBodyPartFactory {
+
+        @Override
+        public ResponseBodyPart newResponseBodyPart(ByteBuf buf, boolean last) {
+            return new LazyResponseBodyPart(buf, last);
+        }
+    }
 }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/HttpProtocol.java

@@ -16,7 +16,8 @@
 package org.asynchttpclient.providers.netty.handler;
 
 import static io.netty.handler.codec.http.HttpResponseStatus.*;
-import static org.asynchttpclient.providers.netty.util.HttpUtil.isNTLM;
+import static org.asynchttpclient.providers.netty.util.HttpUtil.*;
+import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpHeaders;
@@ -33,6 +34,7 @@
 import java.util.Map.Entry;
 
 import org.asynchttpclient.AsyncHandler;
+import org.asynchttpclient.AsyncHandler.STATE;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.FluentCaseInsensitiveStringsMap;
 import org.asynchttpclient.HttpResponseBodyPart;
@@ -42,26 +44,25 @@
 import org.asynchttpclient.Realm;
 import org.asynchttpclient.Request;
 import org.asynchttpclient.RequestBuilder;
-import org.asynchttpclient.AsyncHandler.STATE;
 import org.asynchttpclient.filter.FilterContext;
 import org.asynchttpclient.filter.FilterException;
 import org.asynchttpclient.filter.ResponseFilter;
 import org.asynchttpclient.ntlm.NTLMEngine;
 import org.asynchttpclient.ntlm.NTLMEngineException;
-import org.asynchttpclient.spnego.SpnegoEngine;
 import org.asynchttpclient.providers.netty.Callback;
+import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
 import org.asynchttpclient.providers.netty.channel.Channels;
 import org.asynchttpclient.providers.netty.future.NettyResponseFuture;
 import org.asynchttpclient.providers.netty.request.NettyRequestSender;
-import org.asynchttpclient.providers.netty.response.ResponseBodyPart;
 import org.asynchttpclient.providers.netty.response.ResponseHeaders;
 import org.asynchttpclient.providers.netty.response.ResponseStatus;
+import org.asynchttpclient.spnego.SpnegoEngine;
 import org.asynchttpclient.util.AsyncHttpProviderUtils;
 
 final class HttpProtocol extends Protocol {
 
-    public HttpProtocol(Channels channels, AsyncHttpClientConfig config, NettyRequestSender requestSender) {
-        super(channels, config, requestSender);
+    public HttpProtocol(Channels channels, AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig nettyConfig, NettyRequestSender requestSender) {
+        super(channels, config, nettyConfig, requestSender);
     }
 
     private Realm kerberosChallenge(List<String> proxyAuth, Request request, ProxyServer proxyServer, FluentCaseInsensitiveStringsMap headers, Realm realm,
@@ -183,9 +184,9 @@ private void finishUpdate(final NettyResponseFuture<?> future, ChannelHandlerCon
         markAsDone(future, ctx);
     }
 
-    private final boolean updateBodyAndInterrupt(NettyResponseFuture<?> future, AsyncHandler<?> handler, HttpResponseBodyPart c) throws Exception {
-        boolean state = handler.onBodyPartReceived(c) != STATE.CONTINUE;
-        if (c.closeUnderlyingConnection()) {
+    private final boolean updateBodyAndInterrupt(NettyResponseFuture<?> future, AsyncHandler<?> handler, HttpResponseBodyPart bodyPart) throws Exception {
+        boolean state = handler.onBodyPartReceived(bodyPart) != STATE.CONTINUE;
+        if (bodyPart.isUnderlyingConnectionToBeClosed()) {
             future.setKeepAlive(false);
         }
         return state;
@@ -408,8 +409,9 @@ public void handle(final ChannelHandlerContext ctx, final NettyResponseFuture fu
                         }
                     }
 
-                    if (!interrupt && chunk.content().readableBytes() > 0) {
-                        interrupt = updateBodyAndInterrupt(future, handler, new ResponseBodyPart(chunk.content(), last));
+                    ByteBuf buf = chunk.content();
+                    if (!interrupt && buf.readableBytes() > 0) {
+                        interrupt = updateBodyAndInterrupt(future, handler, nettyConfig.getBodyPartFactory().newResponseBodyPart(buf, last));
                     }
 
                     if (interrupt || last) {

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/NettyChannelHandler.java

@@ -30,6 +30,7 @@
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.providers.netty.Callback;
 import org.asynchttpclient.providers.netty.DiscardEvent;
+import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
 import org.asynchttpclient.providers.netty.channel.Channels;
 import org.asynchttpclient.providers.netty.future.NettyResponseFuture;
 import org.asynchttpclient.providers.netty.future.NettyResponseFutures;
@@ -49,13 +50,13 @@ public class NettyChannelHandler extends ChannelInboundHandlerAdapter {
     private final Protocol httpProtocol;
     private final Protocol webSocketProtocol;
 
-    public NettyChannelHandler(AsyncHttpClientConfig config, NettyRequestSender requestSender, Channels channels, AtomicBoolean isClose) {
+    public NettyChannelHandler(AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig nettyConfig, NettyRequestSender requestSender, Channels channels, AtomicBoolean isClose) {
         this.config = config;
         this.requestSender = requestSender;
         this.channels = channels;
         this.closed = isClose;
-        httpProtocol = new HttpProtocol(channels, config, requestSender);
-        webSocketProtocol = new WebSocketProtocol(channels, config, requestSender);
+        httpProtocol = new HttpProtocol(channels, config, nettyConfig, requestSender);
+        webSocketProtocol = new WebSocketProtocol(channels, config, nettyConfig, requestSender);
     }
 
     @Override

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/Protocol.java

@@ -28,6 +28,7 @@
 import org.asynchttpclient.RequestBuilder;
 import org.asynchttpclient.org.jboss.netty.handler.codec.http.CookieDecoder;
 import org.asynchttpclient.providers.netty.Callback;
+import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
 import org.asynchttpclient.providers.netty.channel.Channels;
 import org.asynchttpclient.providers.netty.future.NettyResponseFuture;
 import org.asynchttpclient.providers.netty.request.NettyRequestSender;
@@ -42,10 +43,12 @@ public abstract class Protocol {
     protected final Channels channels;
     protected final AsyncHttpClientConfig config;
     protected final NettyRequestSender requestSender;
+    protected final NettyAsyncHttpProviderConfig nettyConfig;
 
-    public Protocol(Channels channels, AsyncHttpClientConfig config, NettyRequestSender requestSender) {
+    public Protocol(Channels channels, AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig nettyConfig, NettyRequestSender requestSender) {
         this.channels = channels;
         this.config = config;
+        this.nettyConfig = nettyConfig;
         this.requestSender = requestSender;
     }
 

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/WebSocketProtocol.java

@@ -15,7 +15,8 @@
  */
 package org.asynchttpclient.providers.netty.handler;
 
-import static io.netty.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS;
+import static io.netty.handler.codec.http.HttpResponseStatus.*;
+import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpResponse;
@@ -27,16 +28,17 @@
 
 import java.io.IOException;
 
+import org.asynchttpclient.AsyncHandler.STATE;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.HttpResponseHeaders;
 import org.asynchttpclient.HttpResponseStatus;
 import org.asynchttpclient.Request;
-import org.asynchttpclient.AsyncHandler.STATE;
 import org.asynchttpclient.filter.FilterContext;
 import org.asynchttpclient.filter.FilterException;
 import org.asynchttpclient.filter.ResponseFilter;
 import org.asynchttpclient.providers.netty.Constants;
 import org.asynchttpclient.providers.netty.DiscardEvent;
+import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
 import org.asynchttpclient.providers.netty.channel.Channels;
 import org.asynchttpclient.providers.netty.future.NettyResponseFuture;
 import org.asynchttpclient.providers.netty.request.NettyRequestSender;
@@ -54,8 +56,8 @@ final class WebSocketProtocol extends Protocol {
     private static final byte OPCODE_UNKNOWN = -1;
     protected byte pendingOpcode = OPCODE_UNKNOWN;
 
-    public WebSocketProtocol(Channels channels, AsyncHttpClientConfig config, NettyRequestSender requestSender) {
-        super(channels, config, requestSender);
+    public WebSocketProtocol(Channels channels, AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig nettyConfig, NettyRequestSender requestSender) {
+        super(channels, config, nettyConfig, requestSender);
     }
 
     // We don't need to synchronize as replacing the "ws-decoder" will
@@ -154,14 +156,15 @@ public void handle(ChannelHandlerContext ctx, NettyResponseFuture future, Object
                         pendingOpcode = OPCODE_BINARY;
                     }
 
-                    if (frame.content() != null && frame.content().readableBytes() > 0) {
-                        ResponseBodyPart rp = new ResponseBodyPart(frame.content(), frame.isFinalFragment());
+                    ByteBuf buf = frame.content();
+                    if (buf != null && buf.readableBytes() > 0) {
+                        ResponseBodyPart rp = nettyConfig.getBodyPartFactory().newResponseBodyPart(buf, frame.isFinalFragment());
                         h.onBodyPartReceived(rp);
 
                         if (pendingOpcode == OPCODE_BINARY) {
                             webSocket.onBinaryFragment(rp.getBodyPartBytes(), frame.isFinalFragment());
                         } else {
-                            webSocket.onTextFragment(frame.content().toString(Constants.UTF8), frame.isFinalFragment());
+                            webSocket.onTextFragment(buf.toString(Constants.UTF8), frame.isFinalFragment());
                         }
                     }
                 }