feat: add stream methods for Page (#1425)

* feat: add stream methods

* add clirr ignore rule

* add comments for stream methods

* fix format

* add tests for stream methods

* modify tests

* add showcase tests

* Revert "add showcase tests"

This reverts commit 6fcfa9b.

* add a integration test for stream methods

* add copyright

* change client builder

* set page token

* remove page token in stream all

* remove page token in stream values

* modify paged request

* modify tests

* create users only once

* delete showcase it

* add comments

* change unit test name

Author: JoeWang1127

gax-java/gax/clirr-ignored-differences.xml

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- see http://www.mojohaus.org/clirr-maven-plugin/examples/ignored-differences.html -->
+<differences>
+  <difference>
+    <!-- add default stream methods to `Page` interface -->
+    <differenceType>7012</differenceType>
+    <className>com/google/api/gax/paging/Page</className>
+    <method>* stream*(*)</method>
+  </difference>
+</differences>

gax-java/gax/src/main/java/com/google/api/gax/paging/Page.java

@@ -29,6 +29,9 @@
  */
 package com.google.api.gax.paging;
 
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
 /**
  * A Page object wraps an API list method response.
  *
@@ -52,12 +55,26 @@ public interface Page<ResourceT> {
   Page<ResourceT> getNextPage();
 
   /**
-   * Returns an iterable that traverses all of the elements of the underlying data source. The data
-   * is fetched lazily page by page, where each page may contain multiple elements. A new page is
+   * Returns an iterable that traverses all the elements of the underlying data source. The data is
+   * fetched lazily page by page, where each page may contain multiple elements. A new page is
    * fetched whenever the elements of any particular page are exhausted.
    */
   Iterable<ResourceT> iterateAll();
 
   /** Returns an iterable over the elements in this page. */
   Iterable<ResourceT> getValues();
+
+  /**
+   * Returns a stream that traverses all the elements of the underlying data source. The data is
+   * fetched lazily page by page, where each page may contain multiple elements. A new page is
+   * fetched whenever the elements of any particular page are exhausted.
+   */
+  default Stream<ResourceT> streamAll() {
+    return StreamSupport.stream(iterateAll().spliterator(), false);
+  }
+
+  /** Returns a stream over the elements in this page. */
+  default Stream<ResourceT> streamValues() {
+    return StreamSupport.stream(getValues().spliterator(), false);
+  }
 }

gax-java/gax/src/test/java/com/google/api/gax/rpc/PagingTest.java

@@ -133,6 +133,54 @@ public void pagedByPage() {
     Truth.assertThat(requestCapture.getAllValues()).containsExactly(0, 2, 4).inOrder();
   }
 
+  @Test
+  public void streamValues_streamIsCorrectPerPage() {
+    ArgumentCaptor<Integer> requestCapture = ArgumentCaptor.forClass(Integer.class);
+    Mockito.when(callIntList.futureCall(requestCapture.capture(), Mockito.any()))
+        .thenReturn(ApiFutures.immediateFuture(Arrays.asList(0, 1, 2)))
+        .thenReturn(ApiFutures.immediateFuture(Arrays.asList(3, 4)))
+        .thenReturn(ApiFutures.immediateFuture(Collections.emptyList()));
+
+    Page<Integer> page =
+        FakeCallableFactory.createPagedCallable(
+                callIntList,
+                PagedCallSettings.newBuilder(new ListIntegersPagedResponseFactory()).build(),
+                clientContext)
+            .call(0)
+            .getPage();
+
+    Truth.assertThat(page.streamValues().count()).isEqualTo(3);
+    Truth.assertThat(page.hasNextPage()).isTrue();
+
+    page = page.getNextPage();
+    Truth.assertThat(page.streamValues().count()).isEqualTo(2);
+    Truth.assertThat(page.hasNextPage()).isTrue();
+
+    page = page.getNextPage();
+    Truth.assertThat(page.streamValues().count()).isEqualTo(0);
+    Truth.assertThat(page.hasNextPage()).isFalse();
+    Truth.assertThat(page.getNextPage()).isNull();
+  }
+
+  @Test
+  public void streamAll_streamIsCorrectInAllPages() {
+    ArgumentCaptor<Integer> requestCapture = ArgumentCaptor.forClass(Integer.class);
+    Mockito.when(callIntList.futureCall(requestCapture.capture(), Mockito.any()))
+        .thenReturn(ApiFutures.immediateFuture(Arrays.asList(0, 1, 2)))
+        .thenReturn(ApiFutures.immediateFuture(Arrays.asList(3, 4)))
+        .thenReturn(ApiFutures.immediateFuture(Collections.emptyList()));
+
+    Page<Integer> page =
+        FakeCallableFactory.createPagedCallable(
+                callIntList,
+                PagedCallSettings.newBuilder(new ListIntegersPagedResponseFactory()).build(),
+                clientContext)
+            .call(0)
+            .getPage();
+
+    Truth.assertThat(page.streamAll().count()).isEqualTo(5);
+  }
+
   @Test
   public void pagedByFixedSizeCollection() {
     ArgumentCaptor<Integer> requestCapture = ArgumentCaptor.forClass(Integer.class);