Step5: Reproduce and fixes #51

Author: aesteve

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

@@ -5,16 +5,22 @@
 import io.reactiverse.awssdk.reactivestreams.ReadStreamPublisher;
 import io.vertx.core.Context;
 import io.vertx.core.Vertx;
-import io.vertx.core.http.*;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientOptions;
+import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.http.HttpHeaders;
+import io.vertx.core.http.HttpMethod;
+import io.vertx.core.http.RequestOptions;
 import software.amazon.awssdk.http.SdkHttpFullResponse;
 import software.amazon.awssdk.http.SdkHttpRequest;
 import software.amazon.awssdk.http.SdkHttpResponse;
 import software.amazon.awssdk.http.async.AsyncExecuteRequest;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
 import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler;
 import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
+import software.amazon.awssdk.utils.StringUtils;
 
-import java.util.Optional;
+import java.net.URI;
 import java.util.concurrent.CompletableFuture;
 
 import static java.util.Objects.requireNonNull;
@@ -90,13 +96,21 @@ void executeOnContext(AsyncExecuteRequest asyncExecuteRequest, CompletableFuture
         }
     }
 
-    private static RequestOptions getRequestOptions(SdkHttpRequest request) {
-        return new RequestOptions()
-            .setHost(request.host())
-            .setPort(request.port())
-            .setURI(request.encodedPath())
-            .setSsl("https".equals(request.protocol()));
-    }
+  private static RequestOptions getRequestOptions(SdkHttpRequest request) {
+    return new RequestOptions()
+      .setHost(request.host())
+      .setPort(request.port())
+      .setURI(createRelativeUri(request.getUri()))
+      .setSsl("https".equals(request.protocol()));
+  }
+
+  private static String createRelativeUri(URI uri) {
+    return (StringUtils.isEmpty(uri.getPath()) ? "/" : uri.getPath()) +
+      // AWS requires query parameters to be encoded as defined by RFC 3986.
+      // see https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+      // uri.toASCIIString() returns the URI encoded in this manner
+      (StringUtils.isEmpty(uri.getQuery()) ? "" : "?" + uri.toASCIIString().split("\\?")[1]);
+  }
 
     @Override
     public void close() {

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

@@ -31,9 +31,12 @@
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.S3Object;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -181,6 +184,28 @@ void downloadImageFromBucketWithoutSettingResponseHandler(Vertx vertx, VertxTest
                 .subscribe(getRes -> ctx.completeNow(), ctx::failNow);
     }
 
+  @Test
+  @Order(8)
+  void listObjectsV2(Vertx vertx, VertxTestContext ctx) throws Exception {
+    final Context originalContext = vertx.getOrCreateContext();
+    final S3AsyncClient s3 = s3(originalContext);
+    single(s3.putObject(b -> putObjectReq(b, "obj1"), AsyncRequestBody.fromString("hello")))
+      .flatMap(putObjectResponse1 -> single(s3.putObject(b -> putObjectReq(b, "obj2"), AsyncRequestBody.fromString("hi"))))
+      .flatMap(putObjectResponse2 -> single(s3.listObjectsV2(VertxS3ClientSpec::listObjectsV2Req))
+        .flatMap(listObjectsV2Response1 -> single(s3.listObjectsV2(b -> listObjectsV2ReqWithContToken(b, listObjectsV2Response1.nextContinuationToken())))
+          .map(listObjectsV2Response2 -> {
+            List<S3Object> allObjects = new ArrayList<>(listObjectsV2Response1.contents());
+            allObjects.addAll(listObjectsV2Response2.contents());
+            return allObjects;
+          })
+        ))
+      .subscribe(allObjects -> {
+        ctx.verify(() -> {
+          assertEquals(3, allObjects.size());
+          ctx.completeNow();
+        });
+      }, ctx::failNow);
+  }
 
     /* Utility methods */
     private static Single<AsyncFile> readFileFromDisk(Vertx vertx) {
@@ -207,4 +232,16 @@ private static GetObjectRequest.Builder downloadImgReq(GetObjectRequest.Builder
         return gor.key(IMG_S3_NAME).bucket(BUCKET_NAME);
     }
 
+  private static ListObjectsV2Request.Builder listObjectsV2Req(ListObjectsV2Request.Builder lovr) {
+    return  lovr.maxKeys(2).bucket(BUCKET_NAME);
+  }
+
+  private static ListObjectsV2Request.Builder listObjectsV2ReqWithContToken(ListObjectsV2Request.Builder lovr, String token) {
+    return lovr.maxKeys(2).bucket(BUCKET_NAME).continuationToken(token);
+  }
+
+  private static PutObjectRequest.Builder putObjectReq(PutObjectRequest.Builder por, String key) {
+    return por.bucket(BUCKET_NAME).key(key);
+  }
+
 }