fix(client): async streaming flakiness (#162)

Author: stainless-app[bot]

openai-java-core/src/main/kotlin/com/openai/core/http/PhantomReachableClosingAsyncStreamResponse.kt

@@ -2,6 +2,7 @@ package com.openai.core.http
 
 import com.openai.core.closeWhenPhantomReachable
 import com.openai.core.http.AsyncStreamResponse.Handler
+import java.util.Optional
 import java.util.concurrent.Executor
 
 internal class PhantomReachableClosingAsyncStreamResponse<T>(
@@ -12,13 +13,30 @@ internal class PhantomReachableClosingAsyncStreamResponse<T>(
     }
 
     override fun subscribe(handler: Handler<T>): AsyncStreamResponse<T> = apply {
-        asyncStreamResponse.subscribe(handler)
+        asyncStreamResponse.subscribe(HandlerReferencingAsyncStreamResponse(handler, this))
     }
 
     override fun subscribe(handler: Handler<T>, executor: Executor): AsyncStreamResponse<T> =
         apply {
-            asyncStreamResponse.subscribe(handler, executor)
+            asyncStreamResponse.subscribe(
+                HandlerReferencingAsyncStreamResponse(handler, this),
+                executor
+            )
         }
 
     override fun close() = asyncStreamResponse.close()
 }
+
+/**
+ * A wrapper around a `Handler` that also references an `AsyncStreamResponse` so that the latter
+ * will not only be phantom reachable and get reclaimed early while the handler itself is reachable
+ * and subscribed to the response.
+ */
+private class HandlerReferencingAsyncStreamResponse<T>(
+    private val handler: Handler<T>,
+    private val asyncStreamResponse: AsyncStreamResponse<T>
+) : Handler<T> {
+    override fun onNext(value: T) = handler.onNext(value)
+
+    override fun onComplete(error: Optional<Throwable>) = handler.onComplete(error)
+}