sse/juc: allow set maxBufferCapacity

Author: MukjepScarlet

retrofit-adapters/sse/juc-flow/src/main/java/retrofit2/adapter/sse/java9/SseJucFlowCallAdapter.kt

@@ -32,6 +32,7 @@ import retrofit2.adapter.sse.internal.EventSourceCallAdapter
 
 internal class SseJucFlowCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
   executor: Executor?,
+  private val maxBufferCapacity: Int,
   retrofit: Retrofit,
   idType: Type,
   typeType: Type,
@@ -55,7 +56,7 @@ internal class SseJucFlowCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
     call: Call<ResponseBody>,
   ): Flow.Publisher<ServerSentEvent<ID, TYPE, DATA>> {
     val delegate = delegate.adapt(call)
-    return object : SubmissionPublisher<ServerSentEvent<ID, TYPE, DATA>>(executor, Flow.defaultBufferSize()) {
+    return object : SubmissionPublisher<ServerSentEvent<ID, TYPE, DATA>>(executor, maxBufferCapacity) {
       override fun subscribe(subscriber: Flow.Subscriber<in ServerSentEvent<ID, TYPE, DATA>>?) {
         super.subscribe(subscriber)
         delegate.subscribe(object : SseCallback<ID, TYPE, DATA> {

retrofit-adapters/sse/juc-flow/src/main/java/retrofit2/adapter/sse/java9/SseJucFlowCallAdapterFactory.kt

@@ -26,12 +26,14 @@ import retrofit2.http.Streaming
 
 class SseJucFlowCallAdapterFactory private constructor(
   private val executor: Executor?,
+  private val maxBufferCapacity: Int,
 ) : CallAdapter.Factory() {
 
   companion object {
     @JvmStatic
     @JvmOverloads
-    fun create(executor: Executor? = null) = SseJucFlowCallAdapterFactory(executor)
+    fun create(executor: Executor? = null, maxBufferCapacity: Int = Flow.defaultBufferSize()) =
+      SseJucFlowCallAdapterFactory(executor, maxBufferCapacity)
   }
 
   override fun get(
@@ -72,6 +74,7 @@ class SseJucFlowCallAdapterFactory private constructor(
 
     return SseJucFlowCallAdapter<Any, Any, Any>(
       executor,
+      maxBufferCapacity,
       retrofit,
       idType,
       typeType,