Consolidated ByteBuf retain calls

nataliakoval · Aug 9, 2013 · 7a0b0ad · 7a0b0ad
1 parent 0cf2814
commit 7a0b0ad
Show file tree

Hide file tree

Showing 6 changed files with 37 additions and 33 deletions.
diff --git a/src/main/java/org/littleshoot/proxy/HttpRelayingHandler.java b/src/main/java/org/littleshoot/proxy/HttpRelayingHandler.java
@@ -7,7 +7,6 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.group.ChannelGroup;
-import io.netty.handler.codec.http.FullHttpResponse;
 import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpObject;
@@ -166,7 +165,6 @@ protected void channelRead0(final ChannelHandlerContext ctx, HttpObject httpObje
                 writeEndBuffer = false;
             }
             else {
-                ((FullHttpResponse) response).content().retain();
                 writeEndBuffer = true;
             }
 
@@ -190,7 +188,6 @@ protected void channelRead0(final ChannelHandlerContext ctx, HttpObject httpObje
         } else {
             log.debug("Processing a chunk");
             final HttpContent chunk = (HttpContent) httpObject;
-            chunk.content().retain();
             if (ProxyUtils.isLastChunk(httpObject)) {
                 readingChunks = false;
                 writeEndBuffer = true;

diff --git a/src/main/java/org/littleshoot/proxy/HttpRequestHandler.java b/src/main/java/org/littleshoot/proxy/HttpRequestHandler.java
@@ -338,6 +338,7 @@ private void processChunk(final HttpObject httpObject) {
         // happen if the client sends a chunk directly after the initial 
         // request.
 
+        chunk.content().retain();
         if (this.currentChannelFuture.channel().isActive()) {
             this.currentChannelFuture.channel().writeAndFlush(chunk);
         }
@@ -400,6 +401,9 @@ private void processRequest(final ChannelHandlerContext ctx,
         final class OnConnect {
             public ChannelFuture onConnect(final ChannelFuture cf) {
                 if (request.getMethod() != HttpMethod.CONNECT) {
+                    if (request instanceof HttpContent) {
+                        ((HttpContent) request).content().retain();
+                    }
                     final ChannelFuture writeFuture = 
                         cf.channel().writeAndFlush(request);
                     writeFuture.addListener(new ChannelFutureListener() {
@@ -615,28 +619,33 @@ private ChannelFuture channelFuture(final String hostAndPort) {
         if(StringUtils.isBlank(hostAndPort)) {
             return currentChannelFuture;
         }
-
-        synchronized (this.externalHostsToChannelFutures) {
-            final Queue<ChannelFuture> futures = 
-                this.externalHostsToChannelFutures.get(hostAndPort);
-            if (futures == null) {
-                return null;
-            }
-            if (futures.isEmpty()) {
-                return null;
-            }
-            final ChannelFuture cf = futures.remove();
-
-            if (cf != null && cf.isSuccess() && 
-                !cf.channel().isActive()) {
-                // In this case, the future successfully connected at one
-                // time, but we're no longer connected. We need to remove the
-                // channel and open a new one.
-                removeProxyToWebConnection(hostAndPort);
-                return null;
-            }
-            return cf;
-        }
+
+        // TODO: Ox noticed that the connection reuse logic here is causing
+        // problems in situations where the client is making concurrent
+        // requests to the same server for multiple resources.  Need to discuss
+        // with afisk what this is doing and how (or if) to fix it.
+        return null;
+//        synchronized (this.externalHostsToChannelFutures) {
+//            final Queue<ChannelFuture> futures = 
+//                this.externalHostsToChannelFutures.get(hostAndPort);
+//            if (futures == null) {
+//                return null;
+//            }
+//            if (futures.isEmpty()) {
+//                return null;
+//            }
+//            final ChannelFuture cf = futures.remove();
+//
+//            if (cf != null && cf.isSuccess() && 
+//                !cf.channel().isActive()) {
+//                // In this case, the future successfully connected at one
+//                // time, but we're no longer connected. We need to remove the
+//                // channel and open a new one.
+//                removeProxyToWebConnection(hostAndPort);
+//                return null;
+//            }
+//            return cf;
+//        }
     }
 
     private void writeConnectResponse(final ChannelHandlerContext ctx,

diff --git a/src/main/java/org/littleshoot/proxy/ProxyHttpRequestEncoder.java b/src/main/java/org/littleshoot/proxy/ProxyHttpRequestEncoder.java
@@ -1,8 +1,6 @@
 package org.littleshoot.proxy;
 
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpObject;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpRequestEncoder;
@@ -73,9 +71,6 @@ protected void encode(ChannelHandlerContext ctx, HttpObject msg,
             final HttpRequest toSend;
             if (transparent) {
                 toSend = request;
-                if (request instanceof FullHttpRequest) {
-                    ((FullHttpRequest) request).content().retain();
-                }
             } else {
                 toSend = ProxyUtils.copyHttpRequest(request, keepProxyFormat);
             }
@@ -85,7 +80,6 @@ protected void encode(ChannelHandlerContext ctx, HttpObject msg,
             //LOG.info("Writing modified request: {}", httpRequestCopy);
             super.encode(ctx, toSend, out);
         } else {
-            ((HttpContent) msg).content().retain();
             super.encode(ctx, msg, out);
         }
     }

diff --git a/src/main/java/org/littleshoot/proxy/ProxyHttpResponse.java b/src/main/java/org/littleshoot/proxy/ProxyHttpResponse.java
@@ -1,6 +1,7 @@
 package org.littleshoot.proxy;
 
 import io.netty.handler.codec.DecoderResult;
+import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpObject;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
@@ -17,6 +18,9 @@ public ProxyHttpResponse(final HttpRequest httpRequest,
         this.httpRequest = httpRequest;
         this.httpResponse = httpResponse;
         this.response = response;
+        if (response instanceof HttpContent) {
+            ((HttpContent) response).content().retain();
+        }
     }
 
     public HttpObject getResponse() {

diff --git a/src/main/resources/log4j.properties b/src/main/resources/log4j.properties
@@ -30,4 +30,4 @@ log4j.appender.TextFile.layout.ConversionPattern=%-6r %d{ISO8601} %-5p [%t] %c{2
 
 log4j.logger.org.apache.http=off
 log4j.logger.org.apache.http.wire=off
-log4j.logger.org.littleshoot.proxy=all
+log4j.logger.org.littleshoot.proxy=off
diff --git a/src/test/resources/log4j.properties b/src/test/resources/log4j.properties
@@ -29,5 +29,5 @@ log4j.appender.TextFile.layout.ConversionPattern=%-6r %d{ISO8601} %-5p [%t] %c{2
 
 #log4j.logger.org.apache.http=DEBUG
 #log4j.logger.org.apache.http.wire=ERROR
-log4j.logger.org.littleshoot.proxy=on
+log4j.logger.org.littleshoot.proxy=off
 #log4j.logger.org.littleshoot.proxy.HttpRelayingHandler=off