Support Async pagination

build-tools/src/main/resources/software/amazon/awssdk/spotbugs-suppressions.xml

@@ -60,4 +60,11 @@
         wrap it in an sdk-specific exception. -->
         <Bug pattern="URF_UNREAD_FIELD,DLS_DEAD_LOCAL_STORE,REC_CATCH_EXCEPTION" />
     </Match>
+
+    <!-- NP_NONNULL_PARAM_VIOLATION false postive - https://github.com/spotbugs/spotbugs/issues/484 -->
+    <Match>
+        <Class name="software.amazon.awssdk.core.pagination.async.SequentialSubscriber" />
+        <Method name="onComplete" />
+        <Bug pattern="NP_NONNULL_PARAM_VIOLATION" />
+    </Match>
 </FindBugsFilter>

codegen/src/main/java/software/amazon/awssdk/codegen/docs/AsyncOperationDocProvider.java

@@ -16,9 +16,9 @@
 package software.amazon.awssdk.codegen.docs;
 
 import java.util.Map;
+import software.amazon.awssdk.codegen.internal.ImmutableMapParameter;
 import software.amazon.awssdk.codegen.model.intermediate.IntermediateModel;
 import software.amazon.awssdk.codegen.model.intermediate.OperationModel;
-import software.amazon.awssdk.core.util.ImmutableMapParameter;
 
 /**
  * Implementations of {@link OperationDocProvider} for async client methods. This implementation is for the typical
@@ -86,10 +86,14 @@ protected void applyThrows(DocumentationBuilder docBuilder) {
      * @return Factories to use for the {@link ClientType#ASYNC} method type.
      */
     static Map<SimpleMethodOverload, Factory> asyncFactories() {
-        return ImmutableMapParameter.of(SimpleMethodOverload.NORMAL, AsyncOperationDocProvider::new,
-                                        SimpleMethodOverload.NO_ARG, AsyncNoArg::new,
-                                        SimpleMethodOverload.FILE, AsyncFile::new,
-                                        SimpleMethodOverload.CONSUMER_BUILDER, AsyncConsumerBuilder::new);
+        return new ImmutableMapParameter.Builder<SimpleMethodOverload, Factory>()
+            .put(SimpleMethodOverload.NORMAL, AsyncOperationDocProvider::new)
+            .put(SimpleMethodOverload.NO_ARG, AsyncNoArg::new)
+            .put(SimpleMethodOverload.FILE, AsyncFile::new)
+            .put(SimpleMethodOverload.CONSUMER_BUILDER, AsyncConsumerBuilder::new)
+            .put(SimpleMethodOverload.PAGINATED, AsyncPaginated::new)
+            .put(SimpleMethodOverload.NO_ARG_PAGINATED, AsyncPaginatedNoArg::new)
+            .build();
     }
 
     /**
@@ -150,4 +154,38 @@ protected void applyParams(DocumentationBuilder docBuilder) {
                              opModel.getInputShape().getC2jName());
         }
     }
+
+    /**
+     * Provider for traditional paginated method that takes in a request object and returns a response object.
+     */
+    private static class AsyncPaginated extends AsyncOperationDocProvider {
+
+        private AsyncPaginated(IntermediateModel model, OperationModel opModel) {
+            super(model, opModel);
+        }
+
+        @Override
+        protected String appendToDescription() {
+            return paginationDocs.getDocsForAsyncOperation();
+        }
+
+        @Override
+        protected void applyReturns(DocumentationBuilder docBuilder) {
+            docBuilder.returns("A custom publisher that can be subscribed to request a stream of response pages.");
+        }
+    }
+
+    /**
+     * Provider for paginated simple method that takes no arguments and creates an empty request object.
+     */
+    private static class AsyncPaginatedNoArg extends AsyncPaginated {
+
+        private AsyncPaginatedNoArg(IntermediateModel model, OperationModel opModel) {
+            super(model, opModel);
+        }
+
+        @Override
+        protected void applyParams(DocumentationBuilder docBuilder) {
+        }
+    }
 }

codegen/src/main/java/software/amazon/awssdk/codegen/docs/DocumentationBuilder.java

@@ -30,6 +30,8 @@
  */
 public final class DocumentationBuilder {
 
+    // TODO This prefix is not suitable for paginated operations. Either remove it for paginated operations
+    // or change the statement to something generic
     private static final String ASYNC_THROWS_PREFIX = "The CompletableFuture returned by this method can be completed " +
                                                       "exceptionally with the following exceptions.";
 

codegen/src/main/java/software/amazon/awssdk/codegen/docs/PaginationDocs.java

@@ -17,13 +17,20 @@
 
 import com.squareup.javapoet.ClassName;
 import com.squareup.javapoet.CodeBlock;
+import com.squareup.javapoet.TypeName;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 import software.amazon.awssdk.codegen.model.intermediate.IntermediateModel;
 import software.amazon.awssdk.codegen.model.intermediate.OperationModel;
 import software.amazon.awssdk.codegen.poet.PoetExtensions;
 import software.amazon.awssdk.codegen.utils.PaginatorUtils;
+import software.amazon.awssdk.core.pagination.async.SequentialSubscriber;
 
 public class PaginationDocs {
 
+    private static final String SUBSCRIBE_METHOD_NAME = "subscribe";
+
     private final OperationModel operationModel;
     private final PoetExtensions poetExtensions;
 
@@ -62,7 +69,7 @@ public String getDocsForSyncResponseClass(ClassName clientInterface) {
                         .add("<p>Represents the output for the {@link $T#$L($T)} operation which is a paginated operation."
                              + " This class is an iterable of {@link $T} that can be used to iterate through all the "
                              + "response pages of the operation.</p>",
-                             clientInterface, getSyncPaginatedMethodName(), requestType(), syncResponsePageType())
+                             clientInterface, getPaginatedMethodName(), requestType(), syncResponsePageType())
                         .add("<p>When the operation is called, an instance of this class is returned.  At this point, "
                              + "no service calls are made yet and so there is no guarantee that the request is valid. "
                              + "As you iterate through the iterable, SDK will start lazily loading response pages by making "
@@ -73,10 +80,54 @@ clientInterface, getSyncPaginatedMethodName(), requestType(), syncResponsePageTy
                         .toString();
     }
 
+    /**
+     * Constructs additional documentation on the async client operation that is appended to the service documentation.
+     */
+    public String getDocsForAsyncOperation() {
+        return CodeBlock.builder()
+                        .add("<p>This is a variant of {@link #$L($T)} operation. "
+                             + "The return type is a custom publisher that can be subscribed to request a stream of response "
+                             + "pages. SDK will internally handle making service calls for you.\n</p>",
+                             operationModel.getMethodName(), requestType())
+                        .add("<p>When the operation is called, an instance of this class is returned.  At this point, "
+                             + "no service calls are made yet and so there is no guarantee that the request is valid. "
+                             + "If there are errors in your request, you will see the failures only after you start streaming "
+                             + "the data. The subscribe method should be called as a request to start streaming data. "
+                             + "For more info, see {@link $T#$L($T)}. Each call to the subscribe method will result in a new "
+                             + "{@link $T} i.e., a new contract to stream data from the starting request.</p>",
+                             getPublisherType(), SUBSCRIBE_METHOD_NAME, getSubscriberType(), getSubscriptionType())
+                        .add(getAsyncCodeSnippets())
+                        .build()
+                        .toString();
+    }
+
+    /**
+     * Constructs javadocs for the generated response classes of a paginated operation in Async client.
+     * @param clientInterface A java poet {@link ClassName} type of the Async client interface
+     */
+    public String getDocsForAsyncResponseClass(ClassName clientInterface) {
+        return CodeBlock.builder()
+                        .add("<p>Represents the output for the {@link $T#$L($T)} operation which is a paginated operation."
+                             + " This class is a type of {@link $T} which can be used to provide a sequence of {@link $T} "
+                             + "response pages as per demand from the subscriber.</p>",
+                             clientInterface, getPaginatedMethodName(), requestType(), getPublisherType(),
+                             syncResponsePageType())
+                        .add("<p>When the operation is called, an instance of this class is returned.  At this point, "
+                             + "no service calls are made yet and so there is no guarantee that the request is valid. "
+                             + "If there are errors in your request, you will see the failures only after you start streaming "
+                             + "the data. The subscribe method should be called as a request to start streaming data. "
+                             + "For more info, see {@link $T#$L($T)}. Each call to the subscribe method will result in a new "
+                             + "{@link $T} i.e., a new contract to stream data from the starting request.</p>",
+                             getPublisherType(), SUBSCRIBE_METHOD_NAME, getSubscriberType(), getSubscriptionType())
+                        .add(getAsyncCodeSnippets())
+                        .build()
+                        .toString();
+    }
+
     private String getSyncCodeSnippets() {
         CodeBlock callOperationOnClient = CodeBlock.builder()
                                            .addStatement("$T responses = client.$L(request)", syncPaginatedResponseType(),
-                                                         getSyncPaginatedMethodName())
+                                                         getPaginatedMethodName())
                                            .build();
 
         return CodeBlock.builder()
@@ -103,20 +154,56 @@ private String getSyncCodeSnippets() {
                         .toString();
     }
 
+    private String getAsyncCodeSnippets() {
+        CodeBlock callOperationOnClient = CodeBlock.builder()
+                                                   .addStatement("$T publisher = client.$L(request)",
+                                                                 asyncPaginatedResponseType(),
+                                                                 getPaginatedMethodName())
+                                                   .build();
+
+        return CodeBlock.builder()
+                        .add("\n\n<p>The following are few ways to use the response class:</p>")
+                        .add("1) Using the forEach helper method",
+                             TypeName.get(SequentialSubscriber.class))
+                        .add(buildCode(CodeBlock.builder()
+                                                .add(callOperationOnClient)
+                                                .add(CodeBlock.builder()
+                                                              .addStatement("CompletableFuture<Void> future = publisher"
+                                                                            + ".forEach(res -> "
+                                                                            + "{ // Do something with the response })")
+                                                              .addStatement("future.get()")
+                                                              .build())
+                                                .build()))
+                        .add("\n\n2) Using a custom subscriber")
+                        .add(buildCode(CodeBlock.builder()
+                                                .add(callOperationOnClient)
+                                                .add("publisher.subscribe(new Subscriber<$T>() {\n\n", syncResponsePageType())
+                                                .addStatement("public void onSubscribe($T subscription) { //... }",
+                                                              getSubscriberType())
+                                                .add("\n\n")
+                                                .addStatement("public void onNext($T response) { //... }", syncResponsePageType())
+                                                .add("});")
+                                                .build()))
+                        .add("As the response is a publisher, it can work well with third party reactive streams implementations "
+                             + "like RxJava2.")
+                        .add(noteAboutSyncNonPaginatedMethod())
+                        .build()
+                        .toString();
+    }
+
     private CodeBlock buildCode(CodeBlock codeSnippet) {
         return CodeBlock.builder()
                         .add("<pre>{@code\n")
                         .add(codeSnippet)
                         .add("}</pre>")
                         .build();
-
     }
 
     /**
      * @return Method name for the sync paginated operation
      */
-    private String getSyncPaginatedMethodName() {
-        return PaginatorUtils.getSyncMethodName(operationModel.getMethodName());
+    private String getPaginatedMethodName() {
+        return PaginatorUtils.getPaginatedMethodName(operationModel.getMethodName());
     }
 
     /**
@@ -140,14 +227,42 @@ private ClassName syncResponsePageType() {
     /**
      * @return A Poet {@link ClassName} for the return type of sync paginated operation.
      */
-    public ClassName syncPaginatedResponseType() {
+    private ClassName syncPaginatedResponseType() {
         return poetExtensions.getResponseClassForPaginatedSyncOperation(operationModel.getOperationName());
     }
 
+    /**
+     * @return A Poet {@link ClassName} for the return type of Async paginated operation.
+     */
+    private ClassName asyncPaginatedResponseType() {
+        return poetExtensions.getResponseClassForPaginatedAsyncOperation(operationModel.getOperationName());
+    }
+
     private CodeBlock noteAboutSyncNonPaginatedMethod() {
         return CodeBlock.builder()
                         .add("\n<p><b>Note: If you prefer to have control on service calls, use the {@link #$L($T)} operation."
                              + "</b></p>", operationModel.getMethodName(), requestType())
                         .build();
     }
+
+    /**
+     * @return A Poet {@link ClassName} for the reactive streams {@link Publisher}.
+     */
+    private ClassName getPublisherType() {
+        return ClassName.get(Publisher.class);
+    }
+
+    /**
+     * @return A Poet {@link ClassName} for the reactive streams {@link Subscriber}.
+     */
+    private ClassName getSubscriberType() {
+        return ClassName.get(Subscriber.class);
+    }
+
+    /**
+     * @return A Poet {@link ClassName} for the reactive streams {@link Subscription}.
+     */
+    private ClassName getSubscriptionType() {
+        return ClassName.get(Subscription.class);
+    }
 }

codegen/src/main/java/software/amazon/awssdk/codegen/docs/SyncOperationDocProvider.java

@@ -204,7 +204,7 @@ protected void applyParams(DocumentationBuilder docBuilder) {
     }
 
     /**
-     * Provider for standard paginated method that that takes in a request object and returns a response object.
+     * Provider for standard paginated method that takes in a request object and returns a response object.
      */
     private static class SyncPaginated extends SyncOperationDocProvider {
 
@@ -214,7 +214,12 @@ private SyncPaginated(IntermediateModel model, OperationModel opModel) {
 
         @Override
         protected String appendToDescription() {
-            return opModel.isPaginated() ? paginationDocs.getDocsForSyncOperation() : null;
+            return paginationDocs.getDocsForSyncOperation();
+        }
+
+        @Override
+        protected void applyReturns(DocumentationBuilder docBuilder) {
+            docBuilder.returns("A custom iterable that can be used to iterate through all the response pages.");
         }
     }
 
@@ -230,7 +235,7 @@ private SyncPaginatedNoArg(IntermediateModel model, OperationModel opModel) {
         @Override
         protected void applyParams(DocumentationBuilder docBuilder) {
             // Link to non-simple method for discoverability
-            docBuilder.see("#%s(%s)", PaginatorUtils.getSyncMethodName(opModel.getMethodName()),
+            docBuilder.see("#%s(%s)", PaginatorUtils.getPaginatedMethodName(opModel.getMethodName()),
                            opModel.getInput().getVariableType());
         }
     }

codegen/src/main/java/software/amazon/awssdk/codegen/emitters/tasks/PaginatorsGeneratorTasks.java

@@ -18,15 +18,18 @@
 import static software.amazon.awssdk.utils.FunctionalUtils.safeFunction;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import software.amazon.awssdk.codegen.emitters.GeneratorTask;
 import software.amazon.awssdk.codegen.emitters.GeneratorTaskParams;
 import software.amazon.awssdk.codegen.emitters.PoetGeneratorTask;
 import software.amazon.awssdk.codegen.model.service.PaginatorDefinition;
 import software.amazon.awssdk.codegen.poet.ClassSpec;
-import software.amazon.awssdk.codegen.poet.paginators.PaginatorResponseClassSpec;
+import software.amazon.awssdk.codegen.poet.paginators.AsyncResponseClassSpec;
+import software.amazon.awssdk.codegen.poet.paginators.SyncResponseClassSpec;
 
 public class PaginatorsGeneratorTasks extends BaseGeneratorTasks {
 
@@ -44,15 +47,25 @@ protected boolean hasTasks() {
 
     @Override
     protected List<GeneratorTask> createTasks() throws Exception {
-        info("Emitting paginator classes");
         return model.getPaginators().entrySet().stream()
                     .filter(entry -> entry.getValue().isValid())
-                    .map(safeFunction(this::createTask))
+                    .flatMap(safeFunction(this::createSyncAndAsyncTasks))
                     .collect(Collectors.toList());
     }
 
-    private GeneratorTask createTask(Map.Entry<String, PaginatorDefinition> entry) throws IOException {
-        ClassSpec classSpec = new PaginatorResponseClassSpec(model, entry.getKey(), entry.getValue());
+    private Stream<GeneratorTask> createSyncAndAsyncTasks(Map.Entry<String, PaginatorDefinition> entry) throws IOException {
+        return Arrays.asList(createSyncTask(entry), createAsyncTask(entry))
+                     .stream();
+    }
+
+    private GeneratorTask createSyncTask(Map.Entry<String, PaginatorDefinition> entry) throws IOException {
+        ClassSpec classSpec = new SyncResponseClassSpec(model, entry.getKey(), entry.getValue());
+
+        return new PoetGeneratorTask(paginatorsClassDir, model.getFileHeader(), classSpec);
+    }
+
+    private GeneratorTask createAsyncTask(Map.Entry<String, PaginatorDefinition> entry) throws IOException {
+        ClassSpec classSpec = new AsyncResponseClassSpec(model, entry.getKey(), entry.getValue());
 
         return new PoetGeneratorTask(paginatorsClassDir, model.getFileHeader(), classSpec);
     }

codegen/src/main/java/software/amazon/awssdk/codegen/poet/PoetExtensions.java

@@ -67,11 +67,22 @@ public ClassName getClientClass(String className) {
      * @return A Poet {@link ClassName} for the response type of a paginated operation in the base service package.
      *
      * Example: If operationName is "ListTables", then the response type of the paginated operation
-     * will be "ListTablesPaginator" class in the base service package.
+     * will be "ListTablesIterable" class.
      */
     @ReviewBeforeRelease("Naming of response shape for paginated APIs")
     public ClassName getResponseClassForPaginatedSyncOperation(String operationName) {
-        return ClassName.get(model.getMetadata().getFullPaginatorsPackageName(), operationName + "Paginator");
+        return ClassName.get(model.getMetadata().getFullPaginatorsPackageName(), operationName + "Iterable");
     }
 
+    /**
+     * @param operationName Name of the operation
+     * @return A Poet {@link ClassName} for the response type of a async paginated operation in the base service package.
+     *
+     * Example: If operationName is "ListTables", then the async response type of the paginated operation
+     * will be "ListTablesPublisher" class.
+     */
+    @ReviewBeforeRelease("Naming of response shape for paginated APIs")
+    public ClassName getResponseClassForPaginatedAsyncOperation(String operationName) {
+        return ClassName.get(model.getMetadata().getFullPaginatorsPackageName(), operationName + "Publisher");
+    }
 }