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

@@ -525,53 +525,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.openai.models.finetuning.jobs.FineTuningJob;
 import com.openai.models.finetuning.jobs.JobListPage;
 
-// As an Iterable:
-JobListPage page = client.fineTuning().jobs().list(params);
+JobListPage page = client.fineTuning().jobs().list();
+
+// Process as an Iterable
 for (FineTuningJob job : page.autoPager()) {
     System.out.println(job);
-};
+}
 
-// As a Stream:
-client.fineTuning().jobs().list(params).autoPager().stream()
+// Process as a Stream
+page.autoPager()
+    .stream()
     .limit(50)
     .forEach(job -> System.out.println(job));
 ```
 
-### Asynchronous
+When using the asynchronous client, the method returns an [`AsyncStreamResponse`](openai-java-core/src/main/kotlin/com/openai/core/http/AsyncStreamResponse.kt):
 
 ```java
-// Using forEach, which returns CompletableFuture<Void>:
-asyncClient.fineTuning().jobs().list(params).autoPager()
-    .forEach(job -> System.out.println(job), executor);
+import com.openai.core.http.AsyncStreamResponse;
+import com.openai.models.finetuning.jobs.FineTuningJob;
+import com.openai.models.finetuning.jobs.JobListPageAsync;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+CompletableFuture<JobListPageAsync> pageFuture = client.async().fineTuning().jobs().list();
+
+pageFuture.thenRun(page -> page.autoPager().subscribe(job -> {
+    System.out.println(job);
+}));
+
+// If you need to handle errors or completion of the stream
+pageFuture.thenRun(page -> page.autoPager().subscribe(new AsyncStreamResponse.Handler<>() {
+    @Override
+    public void onNext(FineTuningJob job) {
+        System.out.println(job);
+    }
+
+    @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(job -> {
+        System.out.println(job);
+    })
+    .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.openai.models.finetuning.jobs.FineTuningJob;
 import com.openai.models.finetuning.jobs.JobListPage;
 
-JobListPage page = client.fineTuning().jobs().list(params);
-while (page != null) {
-    for (FineTuningJob job : page.data()) {
+JobListPage page = client.fineTuning().jobs().list();
+while (true) {
+    for (FineTuningJob job : page.items()) {
         System.out.println(job);
     }
 
-    page = page.getNextPage().orElse(null);
+    if (!page.hasNextPage()) {
+        break;
+    }
+
+    page = page.nextPage();
 }
 ```
 
@@ -654,9 +702,7 @@ Requests time out after 10 minutes by default.
 To set a custom timeout, configure the method call using the `timeout` method:
 
 ```java
-import com.openai.models.ChatModel;
 import com.openai.models.chat.completions.ChatCompletion;
-import com.openai.models.chat.completions.ChatCompletionCreateParams;
 
 ChatCompletion chatCompletion = client.chat().completions().create(
   params, RequestOptions.builder().timeout(Duration.ofSeconds(30)).build()
@@ -907,9 +953,7 @@ ChatCompletion chatCompletion = client.chat().completions().create(params).valid
 Or configure the method call to validate the response using the `responseValidation` method:
 
 ```java
-import com.openai.models.ChatModel;
 import com.openai.models.chat.completions.ChatCompletion;
-import com.openai.models.chat.completions.ChatCompletionCreateParams;
 
 ChatCompletion chatCompletion = client.chat().completions().create(
   params, RequestOptions.builder().responseValidation(true).build()

openai-java-core/src/main/kotlin/com/openai/core/AutoPager.kt

@@ -0,0 +1,21 @@
+// File generated from our OpenAPI spec by Stainless.
+
+package com.openai.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)
+}

openai-java-core/src/main/kotlin/com/openai/core/AutoPagerAsync.kt

@@ -0,0 +1,88 @@
+// File generated from our OpenAPI spec by Stainless.
+
+package com.openai.core
+
+import com.openai.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,
+}

openai-java-core/src/main/kotlin/com/openai/core/Page.kt

@@ -0,0 +1,33 @@
+// File generated from our OpenAPI spec by Stainless.
+
+package com.openai.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>
+}

openai-java-core/src/main/kotlin/com/openai/core/PageAsync.kt

@@ -0,0 +1,35 @@
+// File generated from our OpenAPI spec by Stainless.
+
+package com.openai.core
+
+import java.util.concurrent.CompletableFuture
+
+/**
+ * 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 asynchronously. For
+ * synchronous pagination, see the [Page] interface.
+ */
+interface PageAsync<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(): CompletableFuture<out PageAsync<T>>
+
+    /** Returns the items in this page. */
+    fun items(): List<T>
+}

openai-java-core/src/main/kotlin/com/openai/models/batches/BatchListPage.kt

@@ -2,12 +2,12 @@
 
 package com.openai.models.batches
 
+import com.openai.core.AutoPager
+import com.openai.core.Page
 import com.openai.core.checkRequired
 import com.openai.services.blocking.BatchService
 import java.util.Objects
 import java.util.Optional
-import java.util.stream.Stream
-import java.util.stream.StreamSupport
 import kotlin.jvm.optionals.getOrNull
 
 /** @see [BatchService.list] */
@@ -16,7 +16,7 @@ private constructor(
     private val service: BatchService,
     private val params: BatchListParams,
     private val response: BatchListPageResponse,
-) {
+) : Page<Batch> {
 
     /**
      * Delegates to [BatchListPageResponse], but gracefully handles missing data.
@@ -32,19 +32,16 @@ private constructor(
      */
     fun hasMore(): Optional<Boolean> = response._hasMore().getOptional("has_more")
 
-    fun hasNextPage(): Boolean = data().isNotEmpty()
+    override fun items(): List<Batch> = data()
 
-    fun getNextPageParams(): Optional<BatchListParams> {
-        if (!hasNextPage()) {
-            return Optional.empty()
-        }
+    override fun hasNextPage(): Boolean = items().isNotEmpty()
 
-        return Optional.of(params.toBuilder().after(data().last()._id().getOptional("id")).build())
-    }
+    fun nextPageParams(): BatchListParams =
+        params.toBuilder().after(items().last()._id().getOptional("id")).build()
 
-    fun getNextPage(): Optional<BatchListPage> = getNextPageParams().map { service.list(it) }
+    override fun nextPage(): BatchListPage = service.list(nextPageParams())
 
-    fun autoPager(): AutoPager = AutoPager(this)
+    fun autoPager(): AutoPager<Batch> = AutoPager.from(this)
 
     /** The parameters that were used to request this page. */
     fun params(): BatchListParams = params
@@ -113,25 +110,6 @@ private constructor(
             )
     }
 
-    class AutoPager(private val firstPage: BatchListPage) : Iterable<Batch> {
-
-        override fun iterator(): Iterator<Batch> = iterator {
-            var page = firstPage
-            var index = 0
-            while (true) {
-                while (index < page.data().size) {
-                    yield(page.data()[index++])
-                }
-                page = page.getNextPage().getOrNull() ?: break
-                index = 0
-            }
-        }
-
-        fun stream(): Stream<Batch> {
-            return StreamSupport.stream(spliterator(), false)
-        }
-    }
-
     override fun equals(other: Any?): Boolean {
         if (this === other) {
             return true