feat(client)!: extract auto pagination to shared classes

refactor(client)!: refactor async auto-pagination
refactor(client)!: rename `getNextPage{,Params}` to `nextPage{,Params}`
refactor(client)!: swap `nextPage{,Params}` to return non-optional

# Migration

- If you were referencing the `AutoPager` class on a specific `*Page` or `*PageAsync` type, then you should instead reference the shared `AutoPager` and `AutoPagerAsync` types, under the `core` package
- `AutoPagerAsync` now has different usage. You can call `.subscribe(...)` on the returned object instead to get called back each page item. You can also call `onCompleteFuture()` to get a future that completes when all items have been processed. Finally, you can call `.close()` on the returned object to stop auto-paginating early
- If you were referencing `getNextPage` or `getNextPageParams`:
   - Swap to `nextPage()` and `nextPageParams()`
   - Note that these both now return non-optional types (use `hasNextPage()` before calling these, since they will throw if it's impossible to get another page)

There are examples and further information about pagination in the readme.

Author: stainless-app[bot]

README.md

@@ -215,53 +215,101 @@ The SDK throws custom unchecked exception types:
 
 ## Pagination
 
-For methods that return a paginated list of results, this library provides convenient ways access the results either one page at a time, or item-by-item across all pages.
+The SDK defines methods that return a paginated lists of results. It provides convenient ways to access the results either one page at a time or item-by-item across all pages.
 
 ### Auto-pagination
 
-To iterate through all results across all pages, you can use `autoPager`, which automatically handles fetching more pages for you:
+To iterate through all results across all pages, use the `autoPager()` method, which automatically fetches more pages as needed.
 
-### Synchronous
+When using the synchronous client, the method returns an [`Iterable`](https://docs.oracle.com/javase/8/docs/api/java/lang/Iterable.html)
 
 ```java
 import com.withorb.api.models.Coupon;
 import com.withorb.api.models.CouponListPage;
 
-// As an Iterable:
-CouponListPage page = client.coupons().list(params);
+CouponListPage page = client.coupons().list();
+
+// Process as an Iterable
 for (Coupon coupon : page.autoPager()) {
     System.out.println(coupon);
-};
+}
 
-// As a Stream:
-client.coupons().list(params).autoPager().stream()
+// Process as a Stream
+page.autoPager()
+    .stream()
     .limit(50)
     .forEach(coupon -> System.out.println(coupon));
 ```
 
-### Asynchronous
+When using the asynchronous client, the method returns an [`AsyncStreamResponse`](orb-java-core/src/main/kotlin/com/withorb/api/core/http/AsyncStreamResponse.kt):
 
 ```java
-// Using forEach, which returns CompletableFuture<Void>:
-asyncClient.coupons().list(params).autoPager()
-    .forEach(coupon -> System.out.println(coupon), executor);
+import com.withorb.api.core.http.AsyncStreamResponse;
+import com.withorb.api.models.Coupon;
+import com.withorb.api.models.CouponListPageAsync;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+CompletableFuture<CouponListPageAsync> pageFuture = client.async().coupons().list();
+
+pageFuture.thenRun(page -> page.autoPager().subscribe(coupon -> {
+    System.out.println(coupon);
+}));
+
+// If you need to handle errors or completion of the stream
+pageFuture.thenRun(page -> page.autoPager().subscribe(new AsyncStreamResponse.Handler<>() {
+    @Override
+    public void onNext(Coupon coupon) {
+        System.out.println(coupon);
+    }
+
+    @Override
+    public void onComplete(Optional<Throwable> error) {
+        if (error.isPresent()) {
+            System.out.println("Something went wrong!");
+            throw new RuntimeException(error.get());
+        } else {
+            System.out.println("No more!");
+        }
+    }
+}));
+
+// Or use futures
+pageFuture.thenRun(page -> page.autoPager()
+    .subscribe(coupon -> {
+        System.out.println(coupon);
+    })
+    .onCompleteFuture()
+    .whenComplete((unused, error) -> {
+        if (error != null) {
+            System.out.println("Something went wrong!");
+            throw new RuntimeException(error);
+        } else {
+            System.out.println("No more!");
+        }
+    }));
 ```
 
 ### Manual pagination
 
-If none of the above helpers meet your needs, you can also manually request pages one-by-one. A page of results has a `data()` method to fetch the list of objects, as well as top-level `response` and other methods to fetch top-level data about the page. It also has methods `hasNextPage`, `getNextPage`, and `getNextPageParams` methods to help with pagination.
+To access individual page items and manually request the next page, use the `items()`,
+`hasNextPage()`, and `nextPage()` methods:
 
 ```java
 import com.withorb.api.models.Coupon;
 import com.withorb.api.models.CouponListPage;
 
-CouponListPage page = client.coupons().list(params);
-while (page != null) {
-    for (Coupon coupon : page.data()) {
+CouponListPage page = client.coupons().list();
+while (true) {
+    for (Coupon coupon : page.items()) {
         System.out.println(coupon);
     }
 
-    page = page.getNextPage().orElse(null);
+    if (!page.hasNextPage()) {
+        break;
+    }
+
+    page = page.nextPage();
 }
 ```
 
@@ -338,7 +386,6 @@ To set a custom timeout, configure the method call using the `timeout` method:
 
 ```java
 import com.withorb.api.models.Customer;
-import com.withorb.api.models.CustomerCreateParams;
 
 Customer customer = client.customers().create(
   params, RequestOptions.builder().timeout(Duration.ofSeconds(30)).build()
@@ -587,7 +634,6 @@ Or configure the method call to validate the response using the `responseValidat
 
 ```java
 import com.withorb.api.models.Customer;
-import com.withorb.api.models.CustomerCreateParams;
 
 Customer customer = client.customers().create(
   params, RequestOptions.builder().responseValidation(true).build()

orb-java-client-okhttp/src/main/kotlin/com/withorb/api/client/okhttp/OrbOkHttpClient.kt

@@ -13,6 +13,7 @@ import java.net.Proxy
 import java.time.Clock
 import java.time.Duration
 import java.util.Optional
+import java.util.concurrent.Executor
 import kotlin.jvm.optionals.getOrNull
 
 class OrbOkHttpClient private constructor() {
@@ -47,6 +48,10 @@ class OrbOkHttpClient private constructor() {
 
         fun jsonMapper(jsonMapper: JsonMapper) = apply { clientOptions.jsonMapper(jsonMapper) }
 
+        fun streamHandlerExecutor(streamHandlerExecutor: Executor) = apply {
+            clientOptions.streamHandlerExecutor(streamHandlerExecutor)
+        }
+
         fun clock(clock: Clock) = apply { clientOptions.clock(clock) }
 
         fun headers(headers: Headers) = apply { clientOptions.headers(headers) }

orb-java-client-okhttp/src/main/kotlin/com/withorb/api/client/okhttp/OrbOkHttpClientAsync.kt

@@ -13,6 +13,7 @@ import java.net.Proxy
 import java.time.Clock
 import java.time.Duration
 import java.util.Optional
+import java.util.concurrent.Executor
 import kotlin.jvm.optionals.getOrNull
 
 class OrbOkHttpClientAsync private constructor() {
@@ -47,6 +48,10 @@ class OrbOkHttpClientAsync private constructor() {
 
         fun jsonMapper(jsonMapper: JsonMapper) = apply { clientOptions.jsonMapper(jsonMapper) }
 
+        fun streamHandlerExecutor(streamHandlerExecutor: Executor) = apply {
+            clientOptions.streamHandlerExecutor(streamHandlerExecutor)
+        }
+
         fun clock(clock: Clock) = apply { clientOptions.clock(clock) }
 
         fun headers(headers: Headers) = apply { clientOptions.headers(headers) }

orb-java-core/src/main/kotlin/com/withorb/api/core/AutoPager.kt

@@ -0,0 +1,21 @@
+// File generated from our OpenAPI spec by Stainless.
+
+package com.withorb.api.core
+
+import java.util.stream.Stream
+import java.util.stream.StreamSupport
+
+class AutoPager<T> private constructor(private val firstPage: Page<T>) : Iterable<T> {
+
+    companion object {
+
+        fun <T> from(firstPage: Page<T>): AutoPager<T> = AutoPager(firstPage)
+    }
+
+    override fun iterator(): Iterator<T> =
+        generateSequence(firstPage) { if (it.hasNextPage()) it.nextPage() else null }
+            .flatMap { it.items() }
+            .iterator()
+
+    fun stream(): Stream<T> = StreamSupport.stream(spliterator(), false)
+}

orb-java-core/src/main/kotlin/com/withorb/api/core/AutoPagerAsync.kt

@@ -0,0 +1,88 @@
+// File generated from our OpenAPI spec by Stainless.
+
+package com.withorb.api.core
+
+import com.withorb.api.core.http.AsyncStreamResponse
+import java.util.Optional
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionException
+import java.util.concurrent.Executor
+import java.util.concurrent.atomic.AtomicReference
+
+class AutoPagerAsync<T>
+private constructor(private val firstPage: PageAsync<T>, private val defaultExecutor: Executor) :
+    AsyncStreamResponse<T> {
+
+    companion object {
+
+        fun <T> from(firstPage: PageAsync<T>, defaultExecutor: Executor): AutoPagerAsync<T> =
+            AutoPagerAsync(firstPage, defaultExecutor)
+    }
+
+    private val onCompleteFuture = CompletableFuture<Void?>()
+    private val state = AtomicReference(State.NEW)
+
+    override fun subscribe(handler: AsyncStreamResponse.Handler<T>): AsyncStreamResponse<T> =
+        subscribe(handler, defaultExecutor)
+
+    override fun subscribe(
+        handler: AsyncStreamResponse.Handler<T>,
+        executor: Executor,
+    ): AsyncStreamResponse<T> = apply {
+        // TODO(JDK): Use `compareAndExchange` once targeting JDK 9.
+        check(state.compareAndSet(State.NEW, State.SUBSCRIBED)) {
+            if (state.get() == State.SUBSCRIBED) "Cannot subscribe more than once"
+            else "Cannot subscribe after the response is closed"
+        }
+
+        fun PageAsync<T>.handle(): CompletableFuture<Void?> {
+            if (state.get() == State.CLOSED) {
+                return CompletableFuture.completedFuture(null)
+            }
+
+            items().forEach { handler.onNext(it) }
+            return if (hasNextPage()) nextPage().thenCompose { it.handle() }
+            else CompletableFuture.completedFuture(null)
+        }
+
+        executor.execute {
+            firstPage.handle().whenComplete { _, error ->
+                val actualError =
+                    if (error is CompletionException && error.cause != null) error.cause else error
+                try {
+                    handler.onComplete(Optional.ofNullable(actualError))
+                } finally {
+                    try {
+                        if (actualError == null) {
+                            onCompleteFuture.complete(null)
+                        } else {
+                            onCompleteFuture.completeExceptionally(actualError)
+                        }
+                    } finally {
+                        close()
+                    }
+                }
+            }
+        }
+    }
+
+    override fun onCompleteFuture(): CompletableFuture<Void?> = onCompleteFuture
+
+    override fun close() {
+        val previousState = state.getAndSet(State.CLOSED)
+        if (previousState == State.CLOSED) {
+            return
+        }
+
+        // When the stream is closed, we should always consider it closed. If it closed due
+        // to an error, then we will have already completed the future earlier, and this
+        // will be a no-op.
+        onCompleteFuture.complete(null)
+    }
+}
+
+private enum class State {
+    NEW,
+    SUBSCRIBED,
+    CLOSED,
+}

orb-java-core/src/main/kotlin/com/withorb/api/core/ClientOptions.kt

@@ -10,6 +10,10 @@ import com.withorb.api.core.http.QueryParams
 import com.withorb.api.core.http.RetryingHttpClient
 import java.time.Clock
 import java.util.Optional
+import java.util.concurrent.Executor
+import java.util.concurrent.Executors
+import java.util.concurrent.ThreadFactory
+import java.util.concurrent.atomic.AtomicLong
 import kotlin.jvm.optionals.getOrNull
 
 class ClientOptions
@@ -18,6 +22,7 @@ private constructor(
     @get:JvmName("httpClient") val httpClient: HttpClient,
     @get:JvmName("checkJacksonVersionCompatibility") val checkJacksonVersionCompatibility: Boolean,
     @get:JvmName("jsonMapper") val jsonMapper: JsonMapper,
+    @get:JvmName("streamHandlerExecutor") val streamHandlerExecutor: Executor,
     @get:JvmName("clock") val clock: Clock,
     @get:JvmName("baseUrl") val baseUrl: String,
     @get:JvmName("headers") val headers: Headers,
@@ -63,6 +68,7 @@ private constructor(
         private var httpClient: HttpClient? = null
         private var checkJacksonVersionCompatibility: Boolean = true
         private var jsonMapper: JsonMapper = jsonMapper()
+        private var streamHandlerExecutor: Executor? = null
         private var clock: Clock = Clock.systemUTC()
         private var baseUrl: String = PRODUCTION_URL
         private var headers: Headers.Builder = Headers.builder()
@@ -78,6 +84,7 @@ private constructor(
             httpClient = clientOptions.originalHttpClient
             checkJacksonVersionCompatibility = clientOptions.checkJacksonVersionCompatibility
             jsonMapper = clientOptions.jsonMapper
+            streamHandlerExecutor = clientOptions.streamHandlerExecutor
             clock = clientOptions.clock
             baseUrl = clientOptions.baseUrl
             headers = clientOptions.headers.toBuilder()
@@ -97,6 +104,10 @@ private constructor(
 
         fun jsonMapper(jsonMapper: JsonMapper) = apply { this.jsonMapper = jsonMapper }
 
+        fun streamHandlerExecutor(streamHandlerExecutor: Executor) = apply {
+            this.streamHandlerExecutor = streamHandlerExecutor
+        }
+
         fun clock(clock: Clock) = apply { this.clock = clock }
 
         fun baseUrl(baseUrl: String) = apply { this.baseUrl = baseUrl }
@@ -251,6 +262,20 @@ private constructor(
                 ),
                 checkJacksonVersionCompatibility,
                 jsonMapper,
+                streamHandlerExecutor
+                    ?: Executors.newCachedThreadPool(
+                        object : ThreadFactory {
+
+                            private val threadFactory: ThreadFactory =
+                                Executors.defaultThreadFactory()
+                            private val count = AtomicLong(0)
+
+                            override fun newThread(runnable: Runnable): Thread =
+                                threadFactory.newThread(runnable).also {
+                                    it.name = "orb-stream-handler-thread-${count.getAndIncrement()}"
+                                }
+                        }
+                    ),
                 clock,
                 baseUrl,
                 headers.build(),

orb-java-core/src/main/kotlin/com/withorb/api/core/Page.kt

@@ -0,0 +1,33 @@
+// File generated from our OpenAPI spec by Stainless.
+
+package com.withorb.api.core
+
+/**
+ * An interface representing a single page, with items of type [T], from a paginated endpoint
+ * response.
+ *
+ * Implementations of this interface are expected to request additional pages synchronously. For
+ * asynchronous pagination, see the [PageAsync] interface.
+ */
+interface Page<T> {
+
+    /**
+     * Returns whether there's another page after this one.
+     *
+     * The method generally doesn't make requests so the result depends entirely on the data in this
+     * page. If a significant amount of time has passed between requesting this page and calling
+     * this method, then the result could be stale.
+     */
+    fun hasNextPage(): Boolean
+
+    /**
+     * Returns the page after this one by making another request.
+     *
+     * @throws IllegalStateException if it's impossible to get the next page. This exception is
+     *   avoidable by calling [hasNextPage] first.
+     */
+    fun nextPage(): Page<T>
+
+    /** Returns the items in this page. */
+    fun items(): List<T>
+}