move abstract to delegation

Author: MukjepScarlet

retrofit-adapters/sse/core/src/main/java/retrofit2/adapter/sse/EventSourceCallAdapterFactory.kt

@@ -19,6 +19,7 @@ import java.lang.reflect.ParameterizedType
 import java.lang.reflect.Type
 import retrofit2.CallAdapter
 import retrofit2.Retrofit
+import retrofit2.adapter.sse.internal.EventSourceCallAdapter
 import retrofit2.http.Streaming
 
 object EventSourceCallAdapterFactory : CallAdapter.Factory() {

retrofit-adapters/sse/core/src/main/java/retrofit2/adapter/sse/internal/AbstractSseCallAdapter.kt

@@ -13,19 +13,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package retrofit2.adapter.sse
+package retrofit2.adapter.sse.internal
 
 import java.lang.reflect.Type
 import okhttp3.ResponseBody
 import retrofit2.Call
 import retrofit2.CallAdapter
 import retrofit2.Converter
 import retrofit2.Retrofit
-import retrofit2.adapter.sse.internal.RealEventSource
+import retrofit2.adapter.sse.EventSource
 
 private val EMPTY_ARRAY = emptyArray<Annotation>()
 
-internal class EventSourceCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
+class EventSourceCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
   retrofit: Retrofit,
   private val idType: Type,
   private val typeType: Type,

…t2/adapter/sse/EventSourceCallAdapter.kt …r/sse/internal/EventSourceCallAdapter.ktretrofit-adapters/sse/core/src/main/java/retrofit2/adapter/sse/EventSourceCallAdapter.kt renamed to retrofit-adapters/sse/core/src/main/java/retrofit2/adapter/sse/internal/EventSourceCallAdapter.kt retrofit-adapters/sse/core/src/main/java/retrofit2/adapter/sse/EventSourceCallAdapter.kt renamed to retrofit-adapters/sse/core/src/main/java/retrofit2/adapter/sse/internal/EventSourceCallAdapter.kt

@@ -23,17 +23,29 @@ import java.util.concurrent.ForkJoinPool
 import java.util.concurrent.SubmissionPublisher
 import okhttp3.ResponseBody
 import retrofit2.Call
+import retrofit2.CallAdapter
 import retrofit2.Retrofit
+import retrofit2.adapter.sse.EventSource
 import retrofit2.adapter.sse.ServerSentEvent
-import retrofit2.adapter.sse.internal.AbstractSseCallAdapter
+import retrofit2.adapter.sse.SseCallback
+import retrofit2.adapter.sse.internal.EventSourceCallAdapter
 
 internal class SseJucFlowCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
   executor: Executor?,
   retrofit: Retrofit,
   idType: Type,
   typeType: Type,
   dataType: Type,
-) : AbstractSseCallAdapter<ID, TYPE, DATA, Flow.Publisher<ServerSentEvent<ID, TYPE, DATA>>, SubmissionPublisher<ServerSentEvent<ID, TYPE, DATA>>>(retrofit, idType, typeType, dataType) {
+) : CallAdapter<ResponseBody, Flow.Publisher<ServerSentEvent<ID, TYPE, DATA>>> {
+
+  private val delegate = EventSourceCallAdapter<ID, TYPE, DATA>(
+    retrofit,
+    idType,
+    typeType,
+    dataType,
+  )
+
+  override fun responseType(): Type = delegate.responseType()
 
   private val executor: Executor = executor ?: retrofit.callbackExecutor()
     ?: ForkJoinPool.commonPool().takeIf { ForkJoinPool.getCommonPoolParallelism() > 1 }
@@ -42,34 +54,35 @@ internal class SseJucFlowCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
   override fun adapt(
     call: Call<ResponseBody>,
   ): Flow.Publisher<ServerSentEvent<ID, TYPE, DATA>> {
+    val delegate = delegate.adapt(call)
     return object : SubmissionPublisher<ServerSentEvent<ID, TYPE, DATA>>(executor, Flow.defaultBufferSize()) {
       override fun subscribe(subscriber: Flow.Subscriber<in ServerSentEvent<ID, TYPE, DATA>>?) {
         super.subscribe(subscriber)
-        call.attachEventSourceListener(this)
+        delegate.subscribe(object : SseCallback<ID, TYPE, DATA> {
+          override fun onEvent(
+            eventSource: EventSource<ID, TYPE, DATA>,
+            id: ID?,
+            type: TYPE?,
+            data: DATA,
+          ) {
+            submit(ServerSentEvent(id, type, data))
+          }
+
+          override fun onClosed(eventSource: EventSource<ID, TYPE, DATA>) {
+            close()
+          }
+
+          override fun onFailure(eventSource: EventSource<ID, TYPE, DATA>, t: Throwable?) {
+            closeExceptionally(t ?: RuntimeException()) // TODO exception type
+          }
+        })
       }
 
       override fun close() {
-        call.cancel()
+        delegate.cancel()
         super.close()
       }
     }
   }
 
-  override fun emit(
-    builder: SubmissionPublisher<ServerSentEvent<ID, TYPE, DATA>>,
-    event: ServerSentEvent<ID, TYPE, DATA>,
-  ) {
-    builder.submit(event)
-  }
-
-  override fun close(builder: SubmissionPublisher<ServerSentEvent<ID, TYPE, DATA>>) {
-    builder.close()
-  }
-
-  override fun closeExceptionally(
-    builder: SubmissionPublisher<ServerSentEvent<ID, TYPE, DATA>>,
-    t: Throwable,
-  ) {
-    builder.closeExceptionally(t)
-  }
 }

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

@@ -16,46 +16,63 @@
 package retrofit2.adapter.sse.kotlinx
 
 import java.lang.reflect.Type
-import kotlinx.coroutines.channels.ProducerScope
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
 import okhttp3.ResponseBody
 import retrofit2.Call
+import retrofit2.CallAdapter
 import retrofit2.Retrofit
+import retrofit2.adapter.sse.EventSource
 import retrofit2.adapter.sse.ServerSentEvent
-import retrofit2.adapter.sse.internal.AbstractSseCallAdapter
+import retrofit2.adapter.sse.SseCallback
+import retrofit2.adapter.sse.internal.EventSourceCallAdapter
 
 internal class SseKtxFlowCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
   retrofit: Retrofit,
   idType: Type,
   typeType: Type,
   dataType: Type,
-) : AbstractSseCallAdapter<ID, TYPE, DATA, Flow<ServerSentEvent<ID, TYPE, DATA>>, ProducerScope<ServerSentEvent<ID, TYPE, DATA>>>(retrofit, idType, typeType, dataType) {
+) : CallAdapter<ResponseBody, Flow<ServerSentEvent<ID, TYPE, DATA>>> {
+
+  private val delegate = EventSourceCallAdapter<ID, TYPE, DATA>(
+    retrofit,
+    idType,
+    typeType,
+    dataType,
+  )
+
+  override fun responseType(): Type = delegate.responseType()
 
   override fun adapt(
     call: Call<ResponseBody>,
-  ): Flow<ServerSentEvent<ID, TYPE, DATA>> = callbackFlow {
-    call.attachEventSourceListener(this)
-    awaitClose(call::cancel)
-  }
+  ): Flow<ServerSentEvent<ID, TYPE, DATA>> {
+    val delegate = delegate.adapt(call)
+    return callbackFlow {
+      delegate.subscribe(
+        object : SseCallback<ID, TYPE, DATA> {
+          override fun onEvent(
+            eventSource: EventSource<ID, TYPE, DATA>,
+            id: ID?,
+            type: TYPE?,
+            data: DATA,
+          ) {
+            trySendBlocking(ServerSentEvent(id, type, data))
+          }
 
-  override fun emit(
-    builder: ProducerScope<ServerSentEvent<ID, TYPE, DATA>>,
-    event: ServerSentEvent<ID, TYPE, DATA>,
-  ) {
-    builder.trySendBlocking(event)
-  }
+          override fun onClosed(eventSource: EventSource<ID, TYPE, DATA>) {
+            close()
+          }
 
-  override fun close(builder: ProducerScope<ServerSentEvent<ID, TYPE, DATA>>) {
-    builder.close()
-  }
+          override fun onFailure(eventSource: EventSource<ID, TYPE, DATA>, t: Throwable?) {
+            close(t ?: RuntimeException()) // TODO exception type
+          }
+        },
+      )
 
-  override fun closeExceptionally(
-    builder: ProducerScope<ServerSentEvent<ID, TYPE, DATA>>,
-    t: Throwable,
-  ) {
-    builder.close(t)
+      awaitClose(call::cancel)
+    }
   }
+
 }