Added memory leak fix in the ClientToProxyConnection.java based on is…

README.md

@@ -1,5 +1,9 @@
 [![Build Status](https://travis-ci.org/adamfisk/LittleProxy.png?branch=master)](https://travis-ci.org/adamfisk/LittleProxy)
 
+**This branch is used by MAG proxy which the following custom changes:
+- NTLM support 
+- Memory Leak Fix**
+
 LittleProxy is a high performance HTTP proxy written in Java atop Trustin Lee's excellent [Netty](http://netty.io) event-based networking library. It's quite stable, performs well, and is easy to integrate into your projects. 
 
 One option is to clone LittleProxy and run it from the command line. This is as simple as:

pom.xml

@@ -3,7 +3,7 @@
     <groupId>org.littleshoot</groupId>
     <artifactId>littleproxy</artifactId>
     <packaging>jar</packaging>
-    <version>1.1.2-ntlm-20180920</version>
+    <version>1.1.2-ntlm-20190910</version>
     <name>LittleProxy</name>
     <description>
         LittleProxy is a high performance HTTP proxy written in Java and using the Netty networking framework.
@@ -14,9 +14,9 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <github.global.server>github</github.global.server>
-        <netty.version>4.0.44.Final</netty.version>
+        <netty.version>4.1.42.Final</netty.version>
         <slf4j.version>1.7.24</slf4j.version>
-        <java.version>1.7</java.version>
+        <java.version>1.8</java.version>
     </properties>
 
     <organization>

src/main/java/org/littleshoot/proxy/impl/ClientToProxyConnection.java

@@ -21,6 +21,8 @@
 import io.netty.handler.codec.http.HttpVersion;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.handler.traffic.GlobalTrafficShapingHandler;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.ReferenceCounted;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import org.apache.commons.lang3.StringUtils;
@@ -434,6 +436,7 @@ void respond(ProxyToServerConnection serverConnection, HttpFilters filters,
             HttpRequest currentHttpRequest, HttpResponse currentHttpResponse,
             HttpObject httpObject) {
         // we are sending a response to the client, so we are done handling this request
+        ReferenceCountUtil.release(currentRequest);
         this.currentRequest = null;
 
         httpObject = filters.serverToProxyResponse(httpObject);
@@ -784,8 +787,8 @@ protected void exceptionCaught(Throwable cause) {
     private void initChannelPipeline(ChannelPipeline pipeline) {
         LOG.debug("Configuring ChannelPipeline");
 
-        pipeline.addLast("bytesReadMonitor", bytesReadMonitor);
-        pipeline.addLast("bytesWrittenMonitor", bytesWrittenMonitor);
+//        pipeline.addLast("bytesReadMonitor", bytesReadMonitor);
+//        pipeline.addLast("bytesWrittenMonitor", bytesWrittenMonitor);
 
         pipeline.addLast("encoder", new HttpResponseEncoder());
         // We want to allow longer request lines, headers, and chunks
@@ -802,8 +805,8 @@ private void initChannelPipeline(ChannelPipeline pipeline) {
             aggregateContentForFiltering(pipeline, numberOfBytesToBuffer);
         }
 
-        pipeline.addLast("requestReadMonitor", requestReadMonitor);
-        pipeline.addLast("responseWrittenMonitor", responseWrittenMonitor);
+//        pipeline.addLast("requestReadMonitor", requestReadMonitor);
+//        pipeline.addLast("responseWrittenMonitor", responseWrittenMonitor);
 
         pipeline.addLast(
                 "idle",
@@ -871,7 +874,7 @@ private boolean shouldCloseClientConnection(HttpRequest req,
             }
         }
 
-        if (!HttpHeaders.isKeepAlive(req)) {
+        if (req != null && !HttpHeaders.isKeepAlive(req)) {
             LOG.debug("Closing client connection since request is not keep alive: {}", req);
             // Here we simply want to close the connection because the
             // client itself has requested it be closed in the request.
@@ -1272,6 +1275,7 @@ private boolean writeGatewayTimeout(HttpRequest httpRequest) {
      */
     private boolean respondWithShortCircuitResponse(HttpResponse httpResponse) {
         // we are sending a response to the client, so we are done handling this request
+        ReferenceCountUtil.release(currentRequest);
         this.currentRequest = null;
 
         HttpResponse filteredResponse = (HttpResponse) currentFilters.proxyToClientResponse(httpResponse);
@@ -1360,7 +1364,7 @@ protected void setMitming(boolean isMitming) {
      * We track statistics on bytes, requests and responses by adding handlers
      * at the appropriate parts of the pipeline (see initChannelPipeline()).
      **************************************************************************/
-    private final BytesReadMonitor bytesReadMonitor = new BytesReadMonitor() {
+    /*private final BytesReadMonitor bytesReadMonitor = new BytesReadMonitor() {
         @Override
         protected void bytesRead(int numberOfBytes) {
             FlowContext flowContext = flowContext();
@@ -1403,7 +1407,7 @@ protected void responseWritten(HttpResponse httpResponse) {
                         httpResponse);
             }
         }
-    };
+    };*/
 
     private void recordClientConnected() {
         try {

src/main/java/org/littleshoot/proxy/impl/ConnectionFlow.java

@@ -1,5 +1,6 @@
 package org.littleshoot.proxy.impl;
 
+import io.netty.util.ReferenceCounted;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 
@@ -177,11 +178,20 @@ public void operationComplete(
      */
     void succeed() {
         synchronized (connectLock) {
-            serverConnection.getLOG().debug(
-                    "Connection flow completed successfully: {}", currentStep);
-            serverConnection.connectionSucceeded(!suppressInitialRequest);
-            notifyThreadsWaitingForConnection();
-        }
+            try {
+                serverConnection.getLOG().debug(
+                        "Connection flow completed successfully: {}", currentStep);
+                serverConnection.connectionSucceeded(!suppressInitialRequest);
+                notifyThreadsWaitingForConnection();
+            } finally {
+                // we're now done with the initialRequest: it's either been forwarded to the upstream server (HTTP requests), or
+                // completely dropped (HTTPS CONNECTs). if the initialRequest is reference counted (typically because the HttpObjectAggregator is in
+                // the pipeline to generate FullHttpRequests), we need to manually release it to avoid a memory leak.
+                if (serverConnection.getInitialRequest() instanceof ReferenceCounted) {
+                    ((ReferenceCounted)serverConnection.getInitialRequest()).release();
+                }
+            }
+        }   
     }
 
     /**
@@ -199,16 +209,28 @@ void fail(final Throwable cause) {
                     public void operationComplete(Future future)
                             throws Exception {
                         synchronized (connectLock) {
-                            if (!clientConnection.serverConnectionFailed(
-                                    serverConnection,
-                                    lastStateBeforeFailure,
-                                    cause)) {
-                                // the connection to the server failed and we are not retrying, so transition to the
-                                // DISCONNECTED state
-                                serverConnection.become(ConnectionState.DISCONNECTED);
-
-                                // We are not retrying our connection, let anyone waiting for a connection know that we're done
-                                notifyThreadsWaitingForConnection();
+
+                            boolean fallbackToAnotherChainedProxy = false;
+
+                            try {
+                                fallbackToAnotherChainedProxy = clientConnection.serverConnectionFailed(
+                                        serverConnection,
+                                        lastStateBeforeFailure,
+                                        cause);
+                            } finally {
+                                // Do not release when there is fallback chained proxy
+                                if (!fallbackToAnotherChainedProxy) {
+                                    if (serverConnection.getInitialRequest() instanceof ReferenceCounted) {
+                                        ((ReferenceCounted)serverConnection.getInitialRequest()).release();
+                                    }
+
+                                    // the connection to the server failed and we are not retrying, so transition to the
+                                    // DISCONNECTED state
+                                    serverConnection.become(ConnectionState.DISCONNECTED);
+
+                                    // We are not retrying our connection, let anyone waiting for a connection know that we're done
+                                    notifyThreadsWaitingForConnection();
+                                }
                             }
                         }
                     }

src/main/java/org/littleshoot/proxy/impl/ProxyConnection.java

@@ -676,7 +676,7 @@ public final void userEventTriggered(ChannelHandlerContext ctx, Object evt)
     /**
      * Utility handler for monitoring bytes read on this connection.
      */
-    @Sharable
+    /*@Sharable
     protected abstract class BytesReadMonitor extends
             ChannelInboundHandlerAdapter {
         @Override
@@ -696,9 +696,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg)
         protected abstract void bytesRead(int numberOfBytes);
     }
 
-    /**
+    *//**
      * Utility handler for monitoring requests read on this connection.
-     */
+     *//*
     @Sharable
     protected abstract class RequestReadMonitor extends
             ChannelInboundHandlerAdapter {
@@ -719,9 +719,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg)
         protected abstract void requestRead(HttpRequest httpRequest);
     }
 
-    /**
+    *//**
      * Utility handler for monitoring responses read on this connection.
-     */
+     *//*
     @Sharable
     protected abstract class ResponseReadMonitor extends
             ChannelInboundHandlerAdapter {
@@ -742,9 +742,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg)
         protected abstract void responseRead(HttpResponse httpResponse);
     }
 
-    /**
+    *//**
      * Utility handler for monitoring bytes written on this connection.
-     */
+     *//*
     @Sharable
     protected abstract class BytesWrittenMonitor extends
             ChannelOutboundHandlerAdapter {
@@ -766,9 +766,9 @@ public void write(ChannelHandlerContext ctx,
         protected abstract void bytesWritten(int numberOfBytes);
     }
 
-    /**
+    *//**
      * Utility handler for monitoring requests written on this connection.
-     */
+     *//*
     @Sharable
     protected abstract class RequestWrittenMonitor extends
             ChannelOutboundHandlerAdapter {
@@ -796,25 +796,25 @@ public void write(ChannelHandlerContext ctx,
             }
         }
 
-        /**
+        *//**
          * Invoked immediately before an HttpRequest is written.
-         */
+         *//*
         protected abstract void requestWriting(HttpRequest httpRequest);
 
-        /**
+        *//**
          * Invoked immediately after an HttpRequest has been sent.
-         */
+         *//*
         protected abstract void requestWritten(HttpRequest httpRequest);
 
-        /**
+        *//**
          * Invoked immediately after an HttpContent has been sent.
-         */
+         *//*
         protected abstract void contentWritten(HttpContent httpContent);
     }
 
-    /**
+    *//**
      * Utility handler for monitoring responses written on this connection.
-     */
+     *//*
     @Sharable
     protected abstract class ResponseWrittenMonitor extends
             ChannelOutboundHandlerAdapter {
@@ -835,7 +835,7 @@ public void write(ChannelHandlerContext ctx,
 
         protected abstract void responseWritten(HttpResponse httpResponse);
     }
-
+*/
     public ChannelHandlerContext getContext() {
         return ctx;
     }

src/main/java/org/littleshoot/proxy/impl/ProxyToServerConnection.java

@@ -354,7 +354,9 @@ void write(Object msg, HttpFilters filters) {
 
     @Override
     void write(Object msg) {
-        LOG.debug("Requested write of {}", msg);
+        //TODO: Need to revert them to DEBUG 
+        LOG.info("Requested write of {}", msg);
+        LOG.info("Message Instance Type - {}", msg.getClass());
 
         if (msg instanceof ReferenceCounted) {
             LOG.debug("Retaining reference counted message");
@@ -383,6 +385,10 @@ void write(Object msg) {
             // already disconnected
             if (isConnecting() || getCurrentState().isDisconnectingOrDisconnected()) {
                 LOG.debug("Connection failed or timed out while waiting to write message to server. Message will be discarded: {}", msg);
+                // release when disconnected.
+                if (initialRequest instanceof ReferenceCounted) {
+                    ((ReferenceCounted)initialRequest).release();
+                }
                 return;
             }
 
@@ -397,9 +403,8 @@ protected void writeHttp(HttpObject httpObject) {
             chainedProxy.filterRequest(httpObject);
         }
         if (httpObject instanceof HttpRequest) {
-            HttpRequest httpRequest = (HttpRequest) httpObject;
             // Remember that we issued this HttpRequest for later
-            currentHttpRequest = httpRequest;
+            currentHttpRequest = (HttpRequest) httpObject;
         }
         super.writeHttp(httpObject);
     }
@@ -1005,8 +1010,8 @@ private void initChannelPipeline(ChannelPipeline pipeline,
             pipeline.addLast("global-traffic-shaping", trafficHandler);
         }
 
-        pipeline.addLast("bytesReadMonitor", bytesReadMonitor);
-        pipeline.addLast("bytesWrittenMonitor", bytesWrittenMonitor);
+//        pipeline.addLast("bytesReadMonitor", bytesReadMonitor);
+//        pipeline.addLast("bytesWrittenMonitor", bytesWrittenMonitor);
 
         pipeline.addLast("encoder", new HttpRequestEncoder());
         pipeline.addLast("decoder", new HeadAwareHttpResponseDecoder(
@@ -1021,8 +1026,8 @@ private void initChannelPipeline(ChannelPipeline pipeline,
             aggregateContentForFiltering(pipeline, numberOfBytesToBuffer);
         }
 
-        pipeline.addLast("responseReadMonitor", responseReadMonitor);
-        pipeline.addLast("requestWrittenMonitor", requestWrittenMonitor);
+//        pipeline.addLast("responseReadMonitor", responseReadMonitor);
+//        pipeline.addLast("requestWrittenMonitor", requestWrittenMonitor);
 
         // Set idle timeout
         pipeline.addLast(
@@ -1123,7 +1128,7 @@ public static InetSocketAddress addressFor(String hostAndPort, DefaultHttpProxyS
      * We track statistics on bytes, requests and responses by adding handlers
      * at the appropriate parts of the pipeline (see initChannelPipeline()).
      **************************************************************************/
-    private final BytesReadMonitor bytesReadMonitor = new BytesReadMonitor() {
+    /*private final BytesReadMonitor bytesReadMonitor = new BytesReadMonitor() {
         @Override
         protected void bytesRead(int numberOfBytes) {
             FullFlowContext flowContext = new FullFlowContext(clientConnection,
@@ -1186,6 +1191,6 @@ protected void contentWritten(HttpContent httpContent) {
                 currentFilters.proxyToServerRequestSent();
             }
         }
-    };
+    };*/
 
 }