Update apache client 5.6 + correct gzip handling (#1143)

* support new gzip

* get entity details from response

* fix gzip test

* unused import

Author: l-trotta

java-client/build.gradle.kts

@@ -186,7 +186,7 @@ dependencies {
 
     // Apache 2.0
     // https://hc.apache.org/httpcomponents-client-ga/
-    api("org.apache.httpcomponents.client5","httpclient5","5.4.4")
+    api("org.apache.httpcomponents.client5","httpclient5","5.6")
 
     // Apache 2.0
     // https://search.maven.org/artifact/com.google.code.findbugs/jsr305

java-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/BasicAsyncResponseConsumer.java

@@ -19,13 +19,11 @@
 
 package co.elastic.clients.transport.rest5_client.low_level;
 
-
 import org.apache.hc.core5.http.ClassicHttpResponse;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.HttpResponse;
 import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
 import org.apache.hc.core5.http.message.BasicClassicHttpResponse;
-import org.apache.hc.core5.http.nio.entity.AbstractBinAsyncEntityConsumer;
 import org.apache.hc.core5.http.nio.support.AbstractAsyncResponseConsumer;
 import org.apache.hc.core5.http.protocol.HttpContext;
 
@@ -39,7 +37,7 @@ class BasicAsyncResponseConsumer extends AbstractAsyncResponseConsumer<ClassicHt
     /**
      * Creates a new instance of this consumer with the provided buffer limit
      */
-    BasicAsyncResponseConsumer(AbstractBinAsyncEntityConsumer consumer) {
+    BasicAsyncResponseConsumer(BufferedByteConsumer consumer) {
         super(consumer);
     }
 

java-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/BufferedByteConsumer.java

@@ -18,21 +18,28 @@
  */
 package co.elastic.clients.transport.rest5_client.low_level;
 
+import org.apache.hc.core5.concurrent.FutureCallback;
 import org.apache.hc.core5.http.ContentTooLongException;
 import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.EntityDetails;
 import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
-import org.apache.hc.core5.http.nio.entity.AbstractBinAsyncEntityConsumer;
+import org.apache.hc.core5.http.nio.AsyncEntityConsumer;
+import org.apache.hc.core5.http.nio.entity.AbstractBinDataConsumer;
 import org.apache.hc.core5.util.ByteArrayBuffer;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 
 import static co.elastic.clients.transport.rest5_client.low_level.Constants.DEFAULT_BUFFER_INITIAL_CAPACITY;
 
-class BufferedByteConsumer extends AbstractBinAsyncEntityConsumer<ByteArrayEntity> {
+public class BufferedByteConsumer extends AbstractBinDataConsumer implements AsyncEntityConsumer<ByteArrayEntity> {
 
-    private volatile ByteArrayBuffer buffer;
     private final int limit;
-    private ContentType contentType;
+    private volatile ByteArrayBuffer buffer;
+    private volatile FutureCallback<ByteArrayEntity> resultCallback;
+    private volatile ContentType contentType;
+    private volatile String contentEncoding;
+    private volatile ByteArrayEntity result;
 
     BufferedByteConsumer(int bufferLimit) {
         super();
@@ -44,8 +51,11 @@ class BufferedByteConsumer extends AbstractBinAsyncEntityConsumer<ByteArrayEntit
     }
 
     @Override
-    protected void streamStart(final ContentType contentType) {
-        this.contentType = contentType;
+    public void streamStart(final EntityDetails entityDetails,
+                            final FutureCallback<ByteArrayEntity> resultCallback) {
+        this.contentType = entityDetails != null ? ContentType.parse(entityDetails.getContentType()) : null;
+        this.contentEncoding = entityDetails != null ? entityDetails.getContentEncoding() : null;
+        this.resultCallback = resultCallback;
     }
 
     @Override
@@ -64,8 +74,25 @@ protected void data(final ByteBuffer src, final boolean endOfStream) throws Cont
     }
 
     @Override
-    protected ByteArrayEntity generateContent() {
-        return new ByteArrayEntity(buffer.toByteArray(), contentType);
+    protected final void completed() throws IOException {
+        result = new ByteArrayEntity(buffer.toByteArray(), contentType, contentEncoding);
+        if (resultCallback != null) {
+            resultCallback.completed(result);
+        }
+        releaseResources();
+    }
+
+    @Override
+    public ByteArrayEntity getContent() {
+        return result;
+    }
+
+    @Override
+    public final void failed(final Exception cause) {
+        if (resultCallback != null) {
+            resultCallback.failed(cause);
+        }
+        releaseResources();
     }
 
     @Override

java-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java

@@ -43,7 +43,6 @@
 import org.apache.hc.core5.http.HttpEntity;
 import org.apache.hc.core5.http.HttpHost;
 import org.apache.hc.core5.http.HttpRequest;
-import org.apache.hc.core5.http.ProtocolException;
 import org.apache.hc.core5.http.message.RequestLine;
 import org.apache.hc.core5.http.nio.AsyncEntityProducer;
 import org.apache.hc.core5.http.nio.AsyncRequestProducer;
@@ -335,13 +334,7 @@ private ResponseOrResponseException convertResponse(InternalRequest request, Nod
 
         HttpEntity entity = httpResponse.getEntity();
         if (entity != null) {
-            Header encoding = null;
-            try {
-                encoding = httpResponse.getHeader(CONTENT_ENCODING);
-            } catch (ProtocolException e) {
-                throw new IOException("Couldn't retrieve content encoding: " + e);
-            }
-            if (encoding != null && "gzip".equals(encoding.getValue())) {
+            if ("gzip".equals(entity.getContentEncoding())) {
                 // Decompress and cleanup response headers
                 httpResponse.setEntity(new GzipDecompressingEntity(entity));
                 httpResponse.removeHeaders(CONTENT_ENCODING);

java-client/src/test/java/co/elastic/clients/transport/rest5_client/low_level/RestClientGzipCompressionTests.java

@@ -22,6 +22,7 @@
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
 import com.sun.net.httpserver.HttpServer;
+import org.apache.hc.client5.http.impl.async.HttpAsyncClientBuilder;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.HttpEntity;
 import org.apache.hc.core5.http.HttpHost;
@@ -122,10 +123,16 @@ private static byte[] readAll(InputStream in) throws IOException {
         return bos.toByteArray();
     }
 
+    // From org.apache.httpcomponents.client5.httpclient5.5.6, the client decompresses responses
+    // by default, so to test the java client's decompression logic, a custom http client with decompression
+    // disabled must be used.
     private Rest5Client createClient(boolean enableCompression) {
         InetSocketAddress address = httpServer.getAddress();
         return Rest5Client.builder(new HttpHost("http", address.getHostString(), address.getPort()))
             .setCompressionEnabled(enableCompression)
+            .setHttpClient(HttpAsyncClientBuilder.create()
+                .disableContentCompression()
+                .build())
             .build();
     }
 
@@ -205,6 +212,9 @@ public void testCompressingClientAsync() throws Exception {
         Rest5Client restClient = Rest5Client.builder(new HttpHost("http", address.getHostString(),
                 address.getPort()))
             .setCompressionEnabled(true)
+            .setHttpClient(HttpAsyncClientBuilder.create()
+                .disableContentCompression()
+                .build())
             .build();
 
         Request request = new Request("POST", "/");