Allow configuration of compression codec.

src/main/java/com/metamx/http/client/HttpClientConfig.java

@@ -24,6 +24,43 @@
  */
 public class HttpClientConfig
 {
+  public enum CompressionCodec
+  {
+    IDENTITY {
+      @Override
+      public String getEncodingString()
+      {
+        return "identity";
+      }
+    },
+    GZIP {
+      @Override
+      public String getEncodingString()
+      {
+        return "gzip";
+      }
+    },
+    DEFLATE {
+      @Override
+      public String getEncodingString()
+      {
+        return "deflate";
+      }
+    };
+
+    /**
+     * Get the header-ified name of this encoding, which should go in "Accept-Encoding" and
+     * "Content-Encoding" headers. This is not just the lowercasing of the enum name, since
+     * we may one day support x- encodings like LZ4, which would likely be an enum named
+     * "LZ4" that has an encoding string like "x-lz4".
+     *
+     * @return encoding name
+     */
+    public abstract String getEncodingString();
+  }
+
+  public static final CompressionCodec DEFAULT_COMPRESSION_CODEC = CompressionCodec.GZIP;
+
   // Default from NioClientSocketChannelFactory.DEFAULT_BOSS_COUNT, which is private:
   private static final int DEFAULT_BOSS_COUNT = 1;
 
@@ -41,14 +78,23 @@ public static Builder builder()
   private final Duration sslHandshakeTimeout;
   private final int bossPoolSize;
   private final int workerPoolSize;
+  private final CompressionCodec compressionCodec;
 
   @Deprecated // Use the builder instead
   public HttpClientConfig(
       int numConnections,
       SSLContext sslContext
   )
   {
-    this(numConnections, sslContext, Duration.ZERO, null, DEFAULT_BOSS_COUNT, DEFAULT_WORKER_COUNT);
+    this(
+        numConnections,
+        sslContext,
+        Duration.ZERO,
+        null,
+        DEFAULT_BOSS_COUNT,
+        DEFAULT_WORKER_COUNT,
+        DEFAULT_COMPRESSION_CODEC
+    );
   }
 
   @Deprecated // Use the builder instead
@@ -58,7 +104,15 @@ public HttpClientConfig(
       Duration readTimeout
   )
   {
-    this(numConnections, sslContext, readTimeout, null, DEFAULT_BOSS_COUNT, DEFAULT_WORKER_COUNT);
+    this(
+        numConnections,
+        sslContext,
+        readTimeout,
+        null,
+        DEFAULT_BOSS_COUNT,
+        DEFAULT_WORKER_COUNT,
+        DEFAULT_COMPRESSION_CODEC
+    );
   }
 
   @Deprecated // Use the builder instead
@@ -69,7 +123,15 @@ public HttpClientConfig(
       Duration sslHandshakeTimeout
   )
   {
-    this(numConnections, sslContext, readTimeout, sslHandshakeTimeout, DEFAULT_BOSS_COUNT, DEFAULT_WORKER_COUNT);
+    this(
+        numConnections,
+        sslContext,
+        readTimeout,
+        sslHandshakeTimeout,
+        DEFAULT_BOSS_COUNT,
+        DEFAULT_WORKER_COUNT,
+        DEFAULT_COMPRESSION_CODEC
+    );
   }
 
   private HttpClientConfig(
@@ -78,7 +140,8 @@ private HttpClientConfig(
       Duration readTimeout,
       Duration sslHandshakeTimeout,
       int bossPoolSize,
-      int workerPoolSize
+      int workerPoolSize,
+      CompressionCodec compressionCodec
   )
   {
     this.numConnections = numConnections;
@@ -87,6 +150,7 @@ private HttpClientConfig(
     this.sslHandshakeTimeout = sslHandshakeTimeout;
     this.bossPoolSize = bossPoolSize;
     this.workerPoolSize = workerPoolSize;
+    this.compressionCodec = compressionCodec;
   }
 
   public int getNumConnections()
@@ -119,6 +183,11 @@ public int getWorkerPoolSize()
     return workerPoolSize;
   }
 
+  public CompressionCodec getCompressionCodec()
+  {
+    return compressionCodec;
+  }
+
   public static class Builder
   {
     private int numConnections = 1;
@@ -127,6 +196,7 @@ public static class Builder
     private Duration sslHandshakeTimeout = null;
     private int bossCount = DEFAULT_BOSS_COUNT;
     private int workerCount = DEFAULT_WORKER_COUNT;
+    private CompressionCodec compressionCodec = DEFAULT_COMPRESSION_CODEC;
 
     private Builder() {}
 
@@ -172,9 +242,23 @@ public Builder withWorkerCount(int workerCount)
       return this;
     }
 
+    public Builder withCompressionCodec(CompressionCodec compressionCodec)
+    {
+      this.compressionCodec = compressionCodec;
+      return this;
+    }
+
     public HttpClientConfig build()
     {
-      return new HttpClientConfig(numConnections, sslContext, readTimeout, sslHandshakeTimeout, bossCount, workerCount);
+      return new HttpClientConfig(
+          numConnections,
+          sslContext,
+          readTimeout,
+          sslHandshakeTimeout,
+          bossCount,
+          workerCount,
+          compressionCodec
+      );
     }
   }
 }

src/main/java/com/metamx/http/client/HttpClientInit.java

@@ -92,8 +92,11 @@ public void stop()
                       config.getSslHandshakeTimeout() == null ? -1 : config.getSslHandshakeTimeout().getMillis()
                   ),
                   new ResourcePoolConfig(config.getNumConnections())
-              )
-          ).withTimer(timer).withReadTimeout(config.getReadTimeout())
+              ),
+              config.getReadTimeout(),
+              config.getCompressionCodec(),
+              timer
+          )
       );
     }
     catch (Exception e) {

src/main/java/com/metamx/http/client/NettyHttpClient.java

@@ -66,23 +66,26 @@ public class NettyHttpClient extends AbstractHttpClient
 
   private final Timer timer;
   private final ResourcePool<String, ChannelFuture> pool;
+  private final HttpClientConfig.CompressionCodec compressionCodec;
   private final Duration defaultReadTimeout;
 
   public NettyHttpClient(
       ResourcePool<String, ChannelFuture> pool
   )
   {
-    this(pool, null, null);
+    this(pool, null, HttpClientConfig.DEFAULT_COMPRESSION_CODEC, null);
   }
 
-  private NettyHttpClient(
+  NettyHttpClient(
       ResourcePool<String, ChannelFuture> pool,
       Duration defaultReadTimeout,
+      HttpClientConfig.CompressionCodec compressionCodec,
       Timer timer
   )
   {
     this.pool = Preconditions.checkNotNull(pool, "pool");
     this.defaultReadTimeout = defaultReadTimeout;
+    this.compressionCodec = Preconditions.checkNotNull(compressionCodec);
     this.timer = timer;
 
     if (defaultReadTimeout != null && defaultReadTimeout.getMillis() > 0) {
@@ -103,12 +106,12 @@ public void stop()
 
   public HttpClient withReadTimeout(Duration readTimeout)
   {
-    return new NettyHttpClient(pool, readTimeout, timer);
+    return new NettyHttpClient(pool, readTimeout, compressionCodec, timer);
   }
 
   public NettyHttpClient withTimer(Timer timer)
   {
-    return new NettyHttpClient(pool, defaultReadTimeout, timer);
+    return new NettyHttpClient(pool, defaultReadTimeout, compressionCodec, timer);
   }
 
   @Override
@@ -155,7 +158,10 @@ public <Intermediate, Final> ListenableFuture<Final> go(
       httpRequest.headers().add(HttpHeaders.Names.HOST, getHost(url));
     }
 
-    httpRequest.headers().set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
+    // If Accept-Encoding is set in the Request, use that. Otherwise use the default from "compressionCodec".
+    if (!headers.containsKey(HttpHeaders.Names.ACCEPT_ENCODING)) {
+      httpRequest.headers().set(HttpHeaders.Names.ACCEPT_ENCODING, compressionCodec.getEncodingString());
+    }
 
     for (Map.Entry<String, Collection<String>> entry : headers.asMap().entrySet()) {
       String key = entry.getKey();

src/test/java/com/metamx/http/client/FriendlyServersTest.java

@@ -47,6 +47,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Tests with servers that are at least moderately well-behaving.
@@ -101,6 +102,64 @@ public void run()
     }
   }
 
+  @Test
+  public void testCompressionCodecConfig() throws Exception
+  {
+    final ExecutorService exec = Executors.newSingleThreadExecutor();
+    final ServerSocket serverSocket = new ServerSocket(0);
+    final AtomicBoolean foundAcceptEncoding = new AtomicBoolean();
+    exec.submit(
+        new Runnable()
+        {
+          @Override
+          public void run()
+          {
+            while (!Thread.currentThread().isInterrupted()) {
+              try (
+                  Socket clientSocket = serverSocket.accept();
+                  BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+                  OutputStream out = clientSocket.getOutputStream()
+              ) {
+                // Read headers
+                String header;
+                while (!(header = in.readLine()).equals("")) {
+                  if (header.equals("Accept-Encoding: identity")) {
+                    foundAcceptEncoding.set(true);
+                  }
+                }
+                out.write("HTTP/1.1 200 OK\r\nContent-Length: 6\r\n\r\nhello!".getBytes(Charsets.UTF_8));
+              }
+              catch (Exception e) {
+                // Suppress
+              }
+            }
+          }
+        }
+    );
+
+    final Lifecycle lifecycle = new Lifecycle();
+    try {
+      final HttpClientConfig config = HttpClientConfig.builder()
+                                                      .withCompressionCodec(HttpClientConfig.CompressionCodec.IDENTITY)
+                                                      .build();
+      final HttpClient client = HttpClientInit.createClient(config, lifecycle);
+      final StatusResponseHolder response = client
+          .go(
+              new Request(HttpMethod.GET, new URL(String.format("http://localhost:%d/", serverSocket.getLocalPort()))),
+              new StatusResponseHandler(Charsets.UTF_8)
+          ).get();
+
+      Assert.assertEquals(200, response.getStatus().getCode());
+      Assert.assertEquals("hello!", response.getContent());
+      Assert.assertTrue(foundAcceptEncoding.get());
+    }
+    finally {
+      exec.shutdownNow();
+      serverSocket.close();
+      lifecycle.stop();
+    }
+  }
+
   @Test
   public void testFriendlySelfSignedHttpsServer() throws Exception
   {