* merge upstream's master

* upgrade to 4.0.0.CR1

Author: aesteve

build.gradle.kts

@@ -1,4 +1,4 @@
-val vertxVersion = "3.9.4"
+val vertxVersion = "4.0.0.CR1"
 val awsSdkVersion = "2.15.23"
 val junit5Version = "5.4.0"
 val logbackVersion = "1.2.3"

src/main/java/io/reactiverse/awssdk/VertxNioAsyncHttpClient.java

@@ -21,7 +21,10 @@
 import software.amazon.awssdk.utils.StringUtils;
 
 import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
 
 import static java.util.Objects.requireNonNull;
 
@@ -66,42 +69,49 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncExecuteRequest)
     void executeOnContext(AsyncExecuteRequest asyncExecuteRequest, CompletableFuture<Void> fut) {
         final SdkHttpRequest request = asyncExecuteRequest.request();
         final SdkAsyncHttpResponseHandler responseHandler = asyncExecuteRequest.responseHandler();
-
-        final HttpMethod method = MethodConverter.awsToVertx(request.method());
         final RequestOptions options = getRequestOptions(request);
-        final HttpClientRequest vRequest = client.request(method, options).setFollowRedirects(true);
-        request.headers().forEach((headerName, headerValues) ->
-                vRequest.putHeader(headerName, String.join(",", headerValues))
-        );
-        vRequest.putHeader(HttpHeaders.CONNECTION, HttpHeaders.KEEP_ALIVE);
-        vRequest.exceptionHandler(error -> {
+        client.request(options, ar -> {
+          if (ar.failed()) {
+            responseHandler.onError(ar.cause());
+            return;
+          }
+          HttpClientRequest vRequest = ar.result();
+          vRequest.onFailure(error -> {
             responseHandler.onError(error);
             fut.completeExceptionally(error);
-        });
-        vRequest.handler(vResponse -> {
+          });
+          vRequest.onSuccess(vResponse -> {
             final SdkHttpFullResponse.Builder builder = SdkHttpResponse.builder()
-                    .statusCode(vResponse.statusCode())
-                    .statusText(vResponse.statusMessage());
+              .statusCode(vResponse.statusCode())
+              .statusText(vResponse.statusMessage());
             vResponse.headers().forEach(e ->
-                    builder.appendHeader(e.getKey(), e.getValue())
+              builder.appendHeader(e.getKey(), e.getValue())
             );
             responseHandler.onHeaders(builder.build());
             responseHandler.onStream(new ReadStreamPublisher<>(vResponse, fut));
-        });
-        final SdkHttpContentPublisher publisher = asyncExecuteRequest.requestContentPublisher();
-        if (publisher != null) {
+          });
+          final SdkHttpContentPublisher publisher = asyncExecuteRequest.requestContentPublisher();
+          if (publisher != null) {
             publisher.subscribe(new HttpClientRequestSubscriber(vRequest));
-        } else {
+          } else {
             vRequest.end();
-        }
+          }
+        });
     }
 
   private static RequestOptions getRequestOptions(SdkHttpRequest request) {
-    return new RequestOptions()
+    RequestOptions options =  new RequestOptions()
+      .setMethod(MethodConverter.awsToVertx(request.method()))
       .setHost(request.host())
       .setPort(request.port())
       .setURI(createRelativeUri(request.getUri()))
+      .setFollowRedirects(true)
       .setSsl("https".equals(request.protocol()));
+    request.headers().forEach((name, values) -> {
+      options.addHeader(name, values.stream().map(s -> (CharSequence)s).collect(Collectors.toList()));
+    });
+    options.addHeader(HttpHeaders.CONNECTION, HttpHeaders.KEEP_ALIVE);
+    return options;
   }
 
   private static String createRelativeUri(URI uri) {

src/test/java/io/reactiverse/awssdk/RetryContextTest.java

@@ -59,7 +59,7 @@ void setupFailServer(Vertx vertx, VertxTestContext ctx) {
                     }
 
                 })
-                .listen(ctx.completing());
+                .listen(ctx.succeedingThenComplete());
     }
 
     @Test

src/test/java/io/reactiverse/awssdk/integration/s3/VertxS3ClientSpec.java

@@ -6,11 +6,17 @@
 import io.reactiverse.awssdk.integration.LocalStackBaseSpec;
 import io.reactiverse.awssdk.reactivestreams.ReadStreamPublisher;
 import io.reactivex.Single;
+import io.vertx.codegen.annotations.Nullable;
+import io.vertx.core.AsyncResult;
 import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.eventbus.MessageProducer;
 import io.vertx.core.file.OpenOptions;
+import io.vertx.core.streams.Pump;
+import io.vertx.core.streams.WriteStream;
 import io.vertx.junit5.Timeout;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
@@ -155,18 +161,50 @@ void downloadImageFromBucket(Vertx vertx, VertxTestContext ctx) throws Exception
     void downloadImageFromBucketToPump(Vertx vertx, VertxTestContext ctx) throws Exception {
         final Context originalContext = vertx.getOrCreateContext();
         final S3AsyncClient s3 = s3(originalContext);
-        final String ebAddress = "s3-forwarded";
-        final MessageProducer<Buffer> producer = vertx.eventBus().sender(ebAddress);
-        final Buffer received = Buffer.buffer();
+        Buffer received = Buffer.buffer();
         AtomicBoolean handlerCalled = new AtomicBoolean(false);
-        VertxAsyncResponseTransformer<GetObjectResponse> transformer = new VertxAsyncResponseTransformer<>(producer);
-        transformer.setResponseHandler(resp -> {
-            handlerCalled.set(true);
-        });
-        vertx.eventBus().<Buffer>consumer(ebAddress, msg -> {
+        VertxAsyncResponseTransformer<GetObjectResponse> transformer = new VertxAsyncResponseTransformer<>(new WriteStream<Buffer>() {
+          @Override
+          public WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
+            return null;
+          }
+
+          @Override
+          public Future<Void> write(Buffer data) {
+            received.appendBuffer(data);
+            return Future.succeededFuture();
+          }
+
+          @Override
+          public void write(Buffer data, Handler<AsyncResult<Void>> handler) {
+            received.appendBuffer(data);
+            handler.handle(null);
+          }
+
+          @Override
+          public void end(Handler<AsyncResult<Void>> handler) {
             assertTrue(handlerCalled.get(), "Response handler should have been called before first bytes are received");
-            received.appendBuffer(msg.body());
             if (received.length() == fileSize) ctx.completeNow();
+            handler.handle(null);
+          }
+
+          @Override
+          public WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) {
+            return null;
+          }
+
+          @Override
+          public boolean writeQueueFull() {
+            return false;
+          }
+
+          @Override
+          public WriteStream<Buffer> drainHandler(@Nullable Handler<Void> handler) {
+            return null;
+          }
+        });
+        transformer.setResponseHandler(resp -> {
+            handlerCalled.set(true);
         });
         single(s3.getObject(VertxS3ClientSpec::downloadImgReq, transformer))
                 .subscribe(getRes -> {}, ctx::failNow);
@@ -175,11 +213,53 @@ void downloadImageFromBucketToPump(Vertx vertx, VertxTestContext ctx) throws Exc
     @Test
     @Order(7)
     void downloadImageFromBucketWithoutSettingResponseHandler(Vertx vertx, VertxTestContext ctx) throws Exception {
-        final Context originalContext = vertx.getOrCreateContext();
-        final S3AsyncClient s3 = s3(originalContext);
-        final String ebAddress = "s3-forwarded";
-        final MessageProducer<Buffer> producer = vertx.eventBus().sender(ebAddress);
-        VertxAsyncResponseTransformer<GetObjectResponse> transformer = new VertxAsyncResponseTransformer<>(producer);
+      final Context originalContext = vertx.getOrCreateContext();
+      final S3AsyncClient s3 = s3(originalContext);
+      final Buffer received = Buffer.buffer();
+      AtomicBoolean handlerCalled = new AtomicBoolean(false);
+      VertxAsyncResponseTransformer<GetObjectResponse> transformer = new VertxAsyncResponseTransformer<>(new WriteStream<Buffer>() {
+          @Override
+          public WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
+            return null;
+          }
+
+          @Override
+          public Future<Void> write(Buffer data) {
+            received.appendBuffer(data);
+            return Future.succeededFuture();
+          }
+
+          @Override
+          public void write(Buffer data, Handler<AsyncResult<Void>> handler) {
+            received.appendBuffer(data);
+            handler.handle(null);
+          }
+
+          @Override
+          public void end(Handler<AsyncResult<Void>> handler) {
+            assertTrue(handlerCalled.get(), "Response handler should have been called before first bytes are received");
+            if (received.length() == fileSize) ctx.completeNow();
+            handler.handle(null);
+          }
+
+          @Override
+          public WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) {
+            return null;
+          }
+
+          @Override
+          public boolean writeQueueFull() {
+            return false;
+          }
+
+          @Override
+          public WriteStream<Buffer> drainHandler(@Nullable Handler<Void> handler) {
+            return null;
+          }
+        });
+        transformer.setResponseHandler(resp -> {
+          handlerCalled.set(true);
+        });
         single(s3.getObject(VertxS3ClientSpec::downloadImgReq, transformer))
                 .subscribe(getRes -> ctx.completeNow(), ctx::failNow);
     }