avoid double producer (#1122) (#1126)

Co-authored-by: Laura Trotta <153528055+l-trotta@users.noreply.github.com>

Author: github-actions[bot]

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

@@ -22,7 +22,7 @@
 import co.elastic.clients.util.NoCopyByteArrayOutputStream;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.io.entity.AbstractHttpEntity;
-import org.apache.hc.core5.http.nio.AsyncDataProducer;
+import org.apache.hc.core5.http.nio.AsyncEntityProducer;
 import org.apache.hc.core5.http.nio.DataStreamChannel;
 
 import java.io.IOException;
@@ -32,20 +32,24 @@
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
 import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * An HTTP entity based on a sequence of byte buffers.
  */
-class MultiBufferEntity extends AbstractHttpEntity implements AsyncDataProducer {
+class MultiBufferEntity extends AbstractHttpEntity implements AsyncEntityProducer {
 
     private final Iterable<ByteBuffer> buffers;
 
     private Iterator<ByteBuffer> iterator;
     private volatile ByteBuffer currentBuffer;
 
+    private final AtomicReference<Exception> exceptionRef;
+
     MultiBufferEntity(Iterable<ByteBuffer> buffers, ContentType contentType) {
         super(contentType,null,true);
         this.buffers = buffers;
+        this.exceptionRef = new AtomicReference<>();
         init();
     }
 
@@ -69,6 +73,13 @@ public boolean isRepeatable() {
         return true;
     }
 
+    @Override
+    public void failed(Exception cause) {
+        if (exceptionRef.compareAndSet(null, cause)) {
+            releaseResources();
+        }
+    }
+
     @Override
     public long getContentLength() {
         // Use chunked encoding

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

@@ -45,6 +45,7 @@
 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;
 import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
 import org.apache.hc.core5.http.nio.support.AsyncRequestBuilder;
@@ -409,7 +410,7 @@ private void performRequestAsync(
             final RequestContext context;
             context = request.createContextForNextAttempt(nodes.next());
             Future<ClassicHttpResponse> futureRef = client.execute(context.requestProducer, context.asyncResponseConsumer, context.context,
-                new FutureCallback<ClassicHttpResponse>() {
+                new FutureCallback<>() {
                     @Override
                     public void completed(ClassicHttpResponse httpResponse) {
                         try {
@@ -877,7 +878,13 @@ private static class RequestContext {
                 .setHeaders(request.httpRequest.getHeaders());
 
             if (request.httpRequest.getEntity() != null) {
-                builder.setEntity(new BasicAsyncEntityProducer(request.httpRequest.getEntity()));
+                HttpEntity entity = request.httpRequest.getEntity();
+                if (entity instanceof AsyncEntityProducer) {
+                    builder.setEntity((AsyncEntityProducer) request.httpRequest.getEntity());
+                }
+                else {
+                    builder.setEntity(new BasicAsyncEntityProducer(entity));
+                }
             }
 
             this.requestProducer = builder.build();