Fixed compression support for h2c protocol

Signed-off-by: Andriy Redko <[email protected]>
opensearch-project · Oct 26, 2022 · 0ba4dbc · 0ba4dbc
1 parent d96da50
commit 0ba4dbc
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 3 deletions.
diff --git a/...rest-high-level/src/test/java/org/opensearch/client/HighLevelRestClientCompressionIT.java b/...rest-high-level/src/test/java/org/opensearch/client/HighLevelRestClientCompressionIT.java
@@ -31,13 +31,22 @@
 
 package org.opensearch.client;
 
+import org.apache.hc.client5.http.classic.methods.HttpGet;
 import org.apache.hc.client5.http.classic.methods.HttpPost;
 import org.apache.hc.client5.http.classic.methods.HttpPut;
+import org.apache.hc.client5.http.entity.GzipCompressingEntity;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.HttpHeaders;
+import org.apache.hc.core5.http.io.entity.StringEntity;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.search.SearchResponse;
 
 import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
 
 import static org.hamcrest.Matchers.equalTo;
 
@@ -62,4 +71,32 @@ public void testCompressesResponseIfRequested() throws IOException {
         assertEquals(SAMPLE_DOCUMENT, searchResponse.getHits().getHits()[0].getSourceAsString());
     }
 
+    /**
+     * The default CloseableHttpAsyncClient does not support compression out of the box (so that applies to RestClient
+     * and RestHighLevelClient). To check the compression works on both sides, crafting the request using CloseableHttpClient
+     * instead which uses compression by default.
+     */
+    public void testCompressesRequest() throws IOException, URISyntaxException {
+        try (CloseableHttpClient client = HttpClients.custom().build()) {
+            final Node node = client().getNodes().iterator().next();
+            final URI baseUri = new URI(node.getHost().toURI());
+
+            final HttpPut index = new HttpPut(baseUri.resolve("/company/_doc/1"));
+            index.setEntity(new GzipCompressingEntity(new StringEntity(SAMPLE_DOCUMENT, ContentType.APPLICATION_JSON)));
+            try (CloseableHttpResponse response = client.execute(index)) {
+                assertThat(response.getCode(), equalTo(201));
+            }
+
+            final HttpGet refresh = new HttpGet(baseUri.resolve("/_refresh"));
+            try (CloseableHttpResponse response = client.execute(refresh)) {
+                assertThat(response.getCode(), equalTo(200));
+            }
+
+            final HttpPost search = new HttpPost(baseUri.resolve("/_search"));
+            index.setEntity(new GzipCompressingEntity(new StringEntity("{}", ContentType.APPLICATION_JSON)));
+            try (CloseableHttpResponse response = client.execute(search)) {
+                assertThat(response.getCode(), equalTo(200));
+            }
+        }
+    }
 }
diff --git a/.../transport-netty4/src/main/java/org/opensearch/http/netty4/Netty4HttpServerTransport.java b/.../transport-netty4/src/main/java/org/opensearch/http/netty4/Netty4HttpServerTransport.java
@@ -413,10 +413,9 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpMessage msg) throws E
                     // If this handler is hit then no upgrade has been attempted and the client is just talking HTTP
                     final ChannelPipeline pipeline = ctx.pipeline();
                     pipeline.addAfter(ctx.name(), "handler", getRequestHandler());
-                    pipeline.replace(this, "aggregator", aggregator);
+                    pipeline.replace(this, "decoder_compress", new HttpContentDecompressor());
 
-                    ch.pipeline().addLast("decoder_compress", new HttpContentDecompressor());
-                    ch.pipeline().addLast("encoder", new HttpResponseEncoder());
+                    pipeline.addAfter("decoder_compress", "aggregator", aggregator);
                     if (handlingSettings.isCompression()) {
                         ch.pipeline()
                             .addAfter("aggregator", "encoder_compress", new HttpContentCompressor(handlingSettings.getCompressionLevel()));