HttpContentDecompressor is misplaced after upgrading pipeline for HTTP tunnelling, close #1583

Author: slandelle

Diff for: client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -345,7 +345,7 @@ public Future<Channel> updatePipelineForHttpTunneling(ChannelPipeline pipeline,
       if (!isSslHandlerConfigured(pipeline)) {
         SslHandler sslHandler = createSslHandler(requestUri.getHost(), requestUri.getExplicitPort());
         whenHanshaked = sslHandler.handshakeFuture();
-        pipeline.addBefore(CHUNKED_WRITER_HANDLER, SSL_HANDLER, sslHandler);
+        pipeline.addBefore(INFLATER_HANDLER, SSL_HANDLER, sslHandler);
       }
       pipeline.addAfter(SSL_HANDLER, HTTP_CLIENT_CODEC, newHttpClientCodec());
 

Diff for: client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java

@@ -99,6 +99,23 @@ public void testNoDirectRequestBodyWithProxy() throws Exception {
     }
   }
 
+  @Test
+  public void testDecompressBodyWithProxy() throws Exception {
+    AsyncHttpClientConfig config = config()
+      .setFollowRedirect(true)
+      .setProxyServer(proxyServer("localhost", port1).build())
+      .setUseInsecureTrustManager(true)
+      .build();
+    try (AsyncHttpClient asyncHttpClient = asyncHttpClient(config)) {
+      String body = "hello world";
+      Response r = asyncHttpClient.executeRequest(post(getTargetUrl2())
+        .setHeader("X-COMPRESS", "true")
+        .setBody(body)).get();
+      assertEquals(r.getStatusCode(), 200);
+      assertEquals(r.getResponseBody(), body);
+    }
+  }
+
   @Test
   public void testPooledConnectionsWithProxy() throws Exception {
 

Diff for: client/src/test/java/org/asynchttpclient/test/EchoHandler.java

@@ -23,12 +23,15 @@
 import javax.servlet.http.Cookie;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Enumeration;
+import java.util.zip.Deflater;
 
-import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
-import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_MD5;
+import static io.netty.handler.codec.http.HttpHeaderNames.*;
+import static io.netty.handler.codec.http.HttpHeaderValues.CHUNKED;
+import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
 
 public class EchoHandler extends AbstractHandler {
 
@@ -115,18 +118,14 @@ public void handle(String pathInContext, Request request, HttpServletRequest htt
       }
     }
 
-    String requestBodyLength = httpRequest.getHeader("X-" + CONTENT_LENGTH);
+    if (httpRequest.getHeader("X-COMPRESS") != null) {
+      byte[] compressed = deflate(IOUtils.toByteArray(httpRequest.getInputStream()));
+      httpResponse.addIntHeader(CONTENT_LENGTH.toString(), compressed.length);
+      httpResponse.addHeader(CONTENT_ENCODING.toString(), DEFLATE.toString());
+      httpResponse.getOutputStream().write(compressed, 0, compressed.length);
 
-    if (requestBodyLength != null) {
-      byte[] requestBodyBytes = IOUtils.toByteArray(httpRequest.getInputStream());
-      int total = requestBodyBytes.length;
-
-      httpResponse.addIntHeader("X-" + CONTENT_LENGTH, total);
-      String md5 = TestUtils.md5(requestBodyBytes, 0, total);
-      httpResponse.addHeader(CONTENT_MD5.toString(), md5);
-
-      httpResponse.getOutputStream().write(requestBodyBytes, 0, total);
     } else {
+      httpResponse.addHeader(TRANSFER_ENCODING.toString(), CHUNKED.toString());
       int size = 16384;
       if (httpRequest.getContentLength() > 0) {
         size = httpRequest.getContentLength();
@@ -148,4 +147,21 @@ public void handle(String pathInContext, Request request, HttpServletRequest htt
     // FIXME don't always close, depends on the test, cf ReactiveStreamsTest
     httpResponse.getOutputStream().close();
   }
+
+  private static byte[] deflate(byte[] input) throws IOException {
+    Deflater compressor = new Deflater();
+    compressor.setLevel(Deflater.BEST_COMPRESSION);
+
+    compressor.setInput(input);
+    compressor.finish();
+
+    try (ByteArrayOutputStream bos = new ByteArrayOutputStream(input.length)) {
+      byte[] buf = new byte[1024];
+      while (!compressor.finished()) {
+        int count = compressor.deflate(buf);
+        bos.write(buf, 0, count);
+      }
+      return bos.toByteArray();
+    }
+  }
 }