Release SDK updates

autorender-java-client-okhttp/src/main/kotlin/io/autorender/client/okhttp/AutorenderOkHttpClient.kt

@@ -9,6 +9,7 @@ import io.autorender.core.ClientOptions
 import io.autorender.core.LogLevel
 import io.autorender.core.Sleeper
 import io.autorender.core.Timeout
+import io.autorender.core.http.AsyncStreamResponse
 import io.autorender.core.http.Headers
 import io.autorender.core.http.HttpClient
 import io.autorender.core.http.ProxyAuthenticator
@@ -18,6 +19,7 @@ import java.net.Proxy
 import java.time.Clock
 import java.time.Duration
 import java.util.Optional
+import java.util.concurrent.Executor
 import java.util.concurrent.ExecutorService
 import javax.net.ssl.HostnameVerifier
 import javax.net.ssl.SSLSocketFactory
@@ -201,6 +203,17 @@ class AutorenderOkHttpClient private constructor() {
          */
         fun jsonMapper(jsonMapper: JsonMapper) = apply { clientOptions.jsonMapper(jsonMapper) }
 
+        /**
+         * The executor to use for running [AsyncStreamResponse.Handler] callbacks.
+         *
+         * Defaults to a dedicated cached thread pool.
+         *
+         * This class takes ownership of the executor and shuts it down, if possible, when closed.
+         */
+        fun streamHandlerExecutor(streamHandlerExecutor: Executor) = apply {
+            clientOptions.streamHandlerExecutor(streamHandlerExecutor)
+        }
+
         /**
          * The interface to use for delaying execution, like during retries.
          *

autorender-java-client-okhttp/src/main/kotlin/io/autorender/client/okhttp/AutorenderOkHttpClientAsync.kt

@@ -9,6 +9,7 @@ import io.autorender.core.ClientOptions
 import io.autorender.core.LogLevel
 import io.autorender.core.Sleeper
 import io.autorender.core.Timeout
+import io.autorender.core.http.AsyncStreamResponse
 import io.autorender.core.http.Headers
 import io.autorender.core.http.HttpClient
 import io.autorender.core.http.ProxyAuthenticator
@@ -18,6 +19,7 @@ import java.net.Proxy
 import java.time.Clock
 import java.time.Duration
 import java.util.Optional
+import java.util.concurrent.Executor
 import java.util.concurrent.ExecutorService
 import javax.net.ssl.HostnameVerifier
 import javax.net.ssl.SSLSocketFactory
@@ -201,6 +203,17 @@ class AutorenderOkHttpClientAsync private constructor() {
          */
         fun jsonMapper(jsonMapper: JsonMapper) = apply { clientOptions.jsonMapper(jsonMapper) }
 
+        /**
+         * The executor to use for running [AsyncStreamResponse.Handler] callbacks.
+         *
+         * Defaults to a dedicated cached thread pool.
+         *
+         * This class takes ownership of the executor and shuts it down, if possible, when closed.
+         */
+        fun streamHandlerExecutor(streamHandlerExecutor: Executor) = apply {
+            clientOptions.streamHandlerExecutor(streamHandlerExecutor)
+        }
+
         /**
          * The interface to use for delaying execution, like during retries.
          *

autorender-java-core/src/main/kotlin/io/autorender/core/AutoPager.kt

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

autorender-java-core/src/main/kotlin/io/autorender/core/AutoPagerAsync.kt

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

autorender-java-core/src/main/kotlin/io/autorender/core/ClientOptions.kt

@@ -3,6 +3,7 @@
 package io.autorender.core
 
 import com.fasterxml.jackson.databind.json.JsonMapper
+import io.autorender.core.http.AsyncStreamResponse
 import io.autorender.core.http.Headers
 import io.autorender.core.http.HttpClient
 import io.autorender.core.http.LoggingHttpClient
@@ -12,6 +13,11 @@ import io.autorender.core.http.RetryingHttpClient
 import java.time.Clock
 import java.time.Duration
 import java.util.Optional
+import java.util.concurrent.Executor
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.ThreadFactory
+import java.util.concurrent.atomic.AtomicLong
 import kotlin.jvm.optionals.getOrNull
 
 /** A class representing the SDK client configuration. */
@@ -41,6 +47,14 @@ private constructor(
      * needs to be overridden.
      */
     @get:JvmName("jsonMapper") val jsonMapper: JsonMapper,
+    /**
+     * The executor to use for running [AsyncStreamResponse.Handler] callbacks.
+     *
+     * Defaults to a dedicated cached thread pool.
+     *
+     * This class takes ownership of the executor and shuts it down, if possible, when closed.
+     */
+    @get:JvmName("streamHandlerExecutor") val streamHandlerExecutor: Executor,
     /**
      * The interface to use for delaying execution, like during retries.
      *
@@ -153,6 +167,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 sleeper: Sleeper? = null
         private var clock: Clock = Clock.systemUTC()
         private var baseUrl: String? = null
@@ -169,6 +184,7 @@ private constructor(
             httpClient = clientOptions.originalHttpClient
             checkJacksonVersionCompatibility = clientOptions.checkJacksonVersionCompatibility
             jsonMapper = clientOptions.jsonMapper
+            streamHandlerExecutor = clientOptions.streamHandlerExecutor
             sleeper = clientOptions.sleeper
             clock = clientOptions.clock
             baseUrl = clientOptions.baseUrl
@@ -211,6 +227,20 @@ private constructor(
          */
         fun jsonMapper(jsonMapper: JsonMapper) = apply { this.jsonMapper = jsonMapper }
 
+        /**
+         * The executor to use for running [AsyncStreamResponse.Handler] callbacks.
+         *
+         * Defaults to a dedicated cached thread pool.
+         *
+         * This class takes ownership of the executor and shuts it down, if possible, when closed.
+         */
+        fun streamHandlerExecutor(streamHandlerExecutor: Executor) = apply {
+            this.streamHandlerExecutor =
+                if (streamHandlerExecutor is ExecutorService)
+                    PhantomReachableExecutorService(streamHandlerExecutor)
+                else streamHandlerExecutor
+        }
+
         /**
          * The interface to use for delaying execution, like during retries.
          *
@@ -427,6 +457,24 @@ private constructor(
          */
         fun build(): ClientOptions {
             val httpClient = checkRequired("httpClient", httpClient)
+            val streamHandlerExecutor =
+                streamHandlerExecutor
+                    ?: PhantomReachableExecutorService(
+                        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 =
+                                            "autorender-stream-handler-thread-${count.getAndIncrement()}"
+                                    }
+                            }
+                        )
+                    )
             val sleeper = sleeper ?: PhantomReachableSleeper(DefaultSleeper())
 
             val headers = Headers.builder()
@@ -464,6 +512,7 @@ private constructor(
                     .build(),
                 checkJacksonVersionCompatibility,
                 jsonMapper,
+                streamHandlerExecutor,
                 sleeper,
                 clock,
                 baseUrl,
@@ -490,6 +539,7 @@ private constructor(
      */
     fun close() {
         httpClient.close()
+        (streamHandlerExecutor as? ExecutorService)?.shutdown()
         sleeper.close()
     }
 }

autorender-java-core/src/main/kotlin/io/autorender/core/Page.kt

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

autorender-java-core/src/main/kotlin/io/autorender/core/PageAsync.kt

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