Paged Flux abstractions

eng/code-quality-reports/src/main/resources/revapi/revapi.json

@@ -84,6 +84,12 @@
         "exampleUseChainInNewApi": ".*com\\.azure\\.messaging\\.eventhubs\\.checkpointstore\\.blob\\..*",
         "justification": "azure-messaging-eventhubs and azure-storage-blob are used in the Event Hubs checkpoint store."
       },
+      {
+        "regex": true,
+        "code": "java\\.missing\\.(oldSuperType|newSuperType)",
+        "new": "class com\\.azure\\.core\\.http\\.rest\\.((PagedFlux<T extends java\\.lang\\.Object>)|(PagedFluxBase<T extends java\\.lang\\.Object.*>)).*",
+        "justification": "azure.core 1.3.0 will introduce a new base type for GAed PagedFlux. Since the base type is not in azure.core <= 1.2.0, revapi is reporting it. This entry should be removed after 1.3.0 release."
+      },
       {
         "regex": true,
         "code": "java\\.missing\\.(oldClass|newClass)",
@@ -93,4 +99,4 @@
       }
     ]
   }
-]
+]

eng/code-quality-reports/src/main/resources/spotbugs/spotbugs-exclude.xml

@@ -707,4 +707,14 @@
     <Bug pattern="CN_IMPLEMENTS_CLONE_BUT_NOT_CLONEABLE"/>
   </Match>
 
+  <Match>
+    <Or>
+      <Class name="com.azure.core.util.paging.PagedFluxCoreJavaDocCodeSnippets"/>
+    </Or>
+    <Bug pattern="NP_LOAD_OF_KNOWN_NULL_VALUE,
+    SIC_INNER_SHOULD_BE_STATIC_ANON,
+    UMAC_UNCALLABLE_METHOD_OF_ANONYMOUS_CLASS,
+    NP_BOOLEAN_RETURN_NULL"/>
+  </Match>
+
 </FindBugsFilter>

sdk/appconfiguration/azure-data-appconfiguration/src/main/java/com/azure/data/appconfiguration/implementation/ConfigurationSettingPage.java

@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 package com.azure.data.appconfiguration.implementation;
 
+import com.azure.core.util.IterableStream;
 import com.azure.data.appconfiguration.models.ConfigurationSetting;
 import com.azure.core.http.rest.Page;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -30,12 +31,12 @@ public String getContinuationToken() {
     }
 
     /**
-     * Gets the list of {@link ConfigurationSetting ConfigurationSettings} on this page.
+     * Gets the iterable stream of {@link ConfigurationSetting ConfigurationSettings} on this page.
      *
-     * @return The list of {@link ConfigurationSetting ConfigurationSettings}.
+     * @return The iterable stream of {@link ConfigurationSetting ConfigurationSettings}.
      */
     @Override
-    public List<ConfigurationSetting> getItems() {
-        return items;
+    public IterableStream<ConfigurationSetting> getElements() {
+        return IterableStream.of(items);
     }
 }

sdk/core/azure-core-test/pom.xml

@@ -38,7 +38,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
-      <version>1.2.0</version> <!-- {x-version-update;com.azure:azure-core;dependency} -->
+      <version>1.3.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
     </dependency>
 
     <dependency>

sdk/core/azure-core-test/src/test/java/com/azure/core/test/implementation/RestProxyWithMockTests.java

@@ -5,7 +5,6 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -36,6 +35,7 @@
 import com.azure.core.test.http.MockHttpResponse;
 import com.azure.core.test.http.NoOpHttpClient;
 import com.azure.core.util.Base64Url;
+import com.azure.core.util.IterableStream;
 import com.fasterxml.jackson.annotation.JsonGetter;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import java.nio.charset.StandardCharsets;
@@ -468,8 +468,8 @@ static class KeyValuePage implements Page<KeyValue> {
         }
 
         @Override
-        public List<KeyValue> getItems() {
-            return items;
+        public IterableStream<KeyValue> getElements() {
+            return new IterableStream<KeyValue>(items);
         }
 
         @Override
@@ -488,8 +488,8 @@ static class ConformingPage<T> implements Page<T> {
         }
 
         @Override
-        public List<T> getItems() {
-            return items;
+        public IterableStream<T> getElements() {
+            return IterableStream.of(items);
         }
 
         @Override
@@ -611,9 +611,8 @@ public void service2getPageSerializes() {
         StepVerifier.create(createService(Service2.class).getPageAsyncSerializes(page))
             .assertNext(response -> {
                 assertEquals(page.getContinuationToken(), response.getContinuationToken());
-                assertNull(response.getItems());
+                assertEquals(0, response.getItems().size());
             })
             .verifyComplete();
     }
-
 }

sdk/core/azure-core/src/main/java/com/azure/core/http/rest/Page.java

@@ -3,26 +3,31 @@
 
 package com.azure.core.http.rest;
 
+import com.azure.core.util.IterableStream;
+import com.azure.core.util.paging.ContinuablePage;
+
+import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * Represents a paginated REST response from the service.
  *
- * @param <T> Type of the listed objects in that response.
+ * @param <T> Type of items in the page response.
  */
-public interface Page<T> {
-
+public interface Page<T> extends ContinuablePage<String, T> {
     /**
-     * Gets a list of items returned from the service.
+     * Get list of elements in the page.
      *
-     * @return A list of items from the service.
-     */
-    List<T> getItems();
-
-    /**
-     * Gets a link to the next page, or {@code null} if there are no more results.
+     * @return the page elements
      *
-     * @return A link to the next page, or {@code null} if there are no more results.
+     * @deprecated use {@link this#getElements()}.
      */
-    String getContinuationToken();
+    @Deprecated
+    default List<T> getItems() {
+        IterableStream<T> iterableStream = this.getElements();
+        return iterableStream == null
+            ? new ArrayList<>()
+            : this.getElements().stream().collect(Collectors.toList());
+    }
 }

sdk/core/azure-core/src/main/java/com/azure/core/http/rest/PagedFlux.java

@@ -6,22 +6,24 @@
 import com.azure.core.http.HttpRequest;
 
 import java.util.stream.Collectors;
+
+import com.azure.core.util.paging.PageRetriever;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.function.Function;
 import java.util.function.Supplier;
 
 /**
- * This class is a flux that can operate on a {@link PagedResponse} and also provides the ability to operate on
- * individual items. When processing the response by page, each response will contain the items in the page as well as
- * the request details like status code and headers.
+ * This type is a Flux that provides the ability to operate on paginated REST responses of type {@link PagedResponse}
+ * and individual items in such pages. When processing the response by page, each response will contain the items
+ * in the page as well as the REST response details like status code and headers.
  *
  * <p>To process one item at a time, simply subscribe to this flux as shown below </p>
  * <p><strong>Code sample</strong></p>
  * {@codesnippet com.azure.core.http.rest.pagedflux.items}
  *
- * <p>To process one page at a time, use {@link #byPage} method as shown below </p>
+ * <p>To process one page at a time, use {@link #byPage()} method as shown below </p>
  * <p><strong>Code sample</strong></p>
  * {@codesnippet com.azure.core.http.rest.pagedflux.pages}
  *
@@ -37,24 +39,23 @@
  * @see Flux
  */
 public class PagedFlux<T> extends PagedFluxBase<T, PagedResponse<T>> {
-
     /**
-     * Creates an instance of {@link PagedFlux} that consists of only a single page of results. The only argument to
-     * this constructor therefore is a supplier that fetches the first (and known-only) page of {@code T}.
+     * Creates an instance of {@link PagedFlux} that consists of only a single page.
+     * This constructor takes a {@code Supplier} that return the single page of {@code T}.
      *
      * <p><strong>Code sample</strong></p>
      * {@codesnippet com.azure.core.http.rest.pagedflux.singlepage.instantiation}
      *
      * @param firstPageRetriever Supplier that retrieves the first page.
      */
     public PagedFlux(Supplier<Mono<PagedResponse<T>>> firstPageRetriever) {
-        super(firstPageRetriever);
+        this(firstPageRetriever, token -> Mono.empty());
     }
 
     /**
-     * Creates an instance of {@link PagedFlux}. The constructor takes in two arguments. The first argument is a
-     * supplier that fetches the first page of {@code T}. The second argument is a function that fetches subsequent
-     * pages of {@code T}
+     * Creates an instance of {@link PagedFlux}. The constructor takes a {@code Supplier} and
+     * {@code Function}. The {@code Supplier} returns the first page of {@code T},
+     * the {@code Function} retrieves subsequent pages of {@code T}.
      *
      * <p><strong>Code sample</strong></p>
      * {@codesnippet com.azure.core.http.rest.pagedflux.instantiation}
@@ -64,21 +65,65 @@ public PagedFlux(Supplier<Mono<PagedResponse<T>>> firstPageRetriever) {
      */
     public PagedFlux(Supplier<Mono<PagedResponse<T>>> firstPageRetriever,
                      Function<String, Mono<PagedResponse<T>>> nextPageRetriever) {
-        super(firstPageRetriever, nextPageRetriever);
+        this(() -> (continuationToken, pageSize) -> continuationToken == null
+            ? firstPageRetriever.get().flux()
+            : nextPageRetriever.apply(continuationToken).flux(), true);
+    }
+
+    /**
+     * Create PagedFlux backed by Page Retriever Function Supplier.
+     *
+     * @param provider the Page Retrieval Provider
+     * @param ignored param is ignored, exists in signature only to avoid conflict with first ctr
+     */
+    private PagedFlux(Supplier<PageRetriever<String, PagedResponse<T>>> provider, boolean ignored) {
+        super(provider, ignored);
+    }
+
+    /**
+     * Creates an instance of {@link PagedFlux} backed by a Page Retriever Supplier (provider).
+     * When invoked provider should return {@link PageRetriever}. The provider will be called for each
+     * Subscription to the PagedFlux instance. The Page Retriever can get called multiple times in serial
+     * fashion, each time after the completion of the Flux returned from the previous invocation.
+     * The final completion signal will be send to the Subscriber when the last Page emitted by the Flux
+     * returned by Page Retriever has {@code null} continuation token.
+     *
+     * The provider is useful mainly in two scenarios:
+     * <ul>
+     * <li> To manage state across multiple call to Page Retrieval within the same Subscription.
+     * <li> To decorate a PagedFlux to produce new PagedFlux.
+     * </ul>
+     *
+     * <p><strong>Decoration sample</strong></p>
+     * {@codesnippet com.azure.core.http.rest.pagedflux.create.decoration}
+     *
+     * @param provider the Page Retrieval Provider
+     * @param <T> The type of items in a {@link PagedResponse}
+     * @return PagedFlux backed by the Page Retriever Function Supplier
+     */
+    public static <T> PagedFlux<T> create(Supplier<PageRetriever<String, PagedResponse<T>>> provider) {
+        return new PagedFlux<>(provider, true);
     }
 
     /**
-     * Maps this PagedFlux instance of T to a PagedFlux instance of type S as per the provided mapper function.
+     * Maps this PagedFlux instance of T to a PagedFlux instance of type S as per the provided mapper
+     * function.
      *
      * @param mapper The mapper function to convert from type T to type S.
      * @param <S> The mapped type.
      * @return A PagedFlux of type S.
+     * @deprecated refer the decoration samples for {@link PagedFlux#create(Supplier)}.
      */
+    @Deprecated
     public <S> PagedFlux<S> mapPage(Function<T, S> mapper) {
-        return new PagedFlux<S>(() -> getFirstPageRetriever().get()
-            .map(mapPagedResponse(mapper)),
-            continuationToken -> getNextPageRetriever().apply(continuationToken)
-                .map(mapPagedResponse(mapper)));
+        Supplier<PageRetriever<String, PagedResponse<S>>> provider = () -> (continuationToken, pageSize) -> {
+            Flux<PagedResponse<T>> flux = (continuationToken == null)
+                ? byPage()
+                : byPage(continuationToken);
+            return flux
+                .map(mapPagedResponse(mapper));
+        };
+        return PagedFlux.create(provider);
     }
 
     private <S> Function<PagedResponse<T>, PagedResponse<S>> mapPagedResponse(Function<T, S> mapper) {