feature: SSE endpoints

gradle/libs.versions.toml

@@ -49,6 +49,7 @@ kotlinx-serialization-core = { module = "org.jetbrains.kotlinx:kotlinx-serializa
 kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinx-serialization" }
 kotlinx-serialization-proto = { module = "org.jetbrains.kotlinx:kotlinx-serialization-protobuf", version.ref = "kotlinx-serialization" }
 okhttp-client = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
+okhttp-sse = { module = "com.squareup.okhttp3:okhttp-sse", version.ref = "okhttp" }
 okhttp-loggingInterceptor = { module = "com.squareup.okhttp3:logging-interceptor", version.ref = "okhttp" }
 okhttp-mockwebserver = { module = "com.squareup.okhttp3:mockwebserver", version.ref = "okhttp" }
 junit = { module = "junit:junit", version = "4.13.2" }

retrofit-adapters/sse/core/build.gradle

@@ -0,0 +1,21 @@
+apply plugin: 'java-library'
+apply plugin: 'org.jetbrains.kotlin.jvm'
+apply plugin: 'com.vanniktech.maven.publish'
+apply plugin: 'org.jetbrains.dokka'
+
+dependencies {
+  api projects.retrofit
+  api libs.okhttp.sse
+  compileOnly libs.findBugsAnnotations
+
+  testImplementation libs.junit
+  testImplementation libs.truth
+  testImplementation libs.okhttp.mockwebserver
+  testImplementation projects.retrofitConverters.scalars
+}
+
+jar {
+  manifest {
+    attributes 'Automatic-Module-Name': 'retrofit2.adapter.sse-core'
+  }
+}

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

@@ -0,0 +1,27 @@
+/*
+ * Copyright (C) 2017 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package retrofit2.adapter.sse
+
+import okhttp3.Request
+
+interface EventSource<ID : Any, TYPE : Any, DATA : Any> {
+
+  fun request(): Request
+
+  fun cancel()
+
+  fun subscribe(callback: SseCallback<ID, TYPE, DATA>)
+}

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

@@ -0,0 +1,63 @@
+/*
+ * Copyright (C) 2017 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package retrofit2.adapter.sse
+
+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.GET
+import retrofit2.http.Streaming
+
+object EventSourceCallAdapterFactory : CallAdapter.Factory() {
+
+  override fun get(
+    returnType: Type,
+    annotations: Array<out Annotation?>,
+    retrofit: Retrofit,
+  ): CallAdapter<*, *>? {
+    if (getRawType(returnType) != EventSource::class.java) {
+      return null
+    }
+
+    if (returnType !is ParameterizedType) {
+      error(
+        "EventSource return type must be parameterized as EventSource<ID, TYPE, DATA>" +
+          " or EventSource<? extends ID, ? extends TYPE, ? extends DATA>",
+      )
+    }
+
+    if (annotations.none { it is Streaming }) {
+      error("SSE endpoint must be annotated with @Streaming")
+    }
+
+    if (annotations.none { it is GET }) {
+      error("SSE endpoint must use @GET method")
+    }
+
+    val idType = getParameterUpperBound(0, returnType)
+    val typeType = getParameterUpperBound(1, returnType)
+    val dataType = getParameterUpperBound(2, returnType)
+
+    return EventSourceCallAdapter<Any, Any, Any>(
+      retrofit,
+      idType,
+      typeType,
+      dataType,
+    )
+  }
+}

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright (C) 2017 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package retrofit2.adapter.sse
+
+/**
+ * A server-sent event.
+ */
+data class ServerSentEvent<ID : Any, TYPE : Any, DATA : Any>(
+  @get:JvmName("id")
+  val id: ID?,
+
+  @get:JvmName("type")
+  val type: TYPE?,
+
+  @get:JvmName("data")
+  val data: DATA,
+)

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright (C) 2017 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package retrofit2.adapter.sse
+
+interface SseCallback<ID : Any, TYPE : Any, DATA : Any> {
+
+  fun onOpen(eventSource: EventSource<ID, TYPE, DATA>) {
+  }
+
+  fun onEvent(eventSource: EventSource<ID, TYPE, DATA>, id: ID?, type: TYPE?, data: DATA) {
+  }
+
+  fun onClosed(eventSource: EventSource<ID, TYPE, DATA>) {
+  }
+
+  fun onFailure(eventSource: EventSource<ID, TYPE, DATA>, t: Throwable?) {
+  }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright (C) 2017 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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.EventSource
+
+private val EMPTY_ARRAY = emptyArray<Annotation>()
+
+class EventSourceCallAdapter<ID : Any, TYPE : Any, DATA : Any>(
+  retrofit: Retrofit,
+  private val idType: Type,
+  private val typeType: Type,
+  private val dataType: Type,
+) : CallAdapter<ResponseBody, EventSource<ID, TYPE, DATA>> {
+
+  private val idConverter: Converter<ResponseBody, ID> = retrofit.responseBodyConverter(idType, EMPTY_ARRAY)
+  private val typeConverter: Converter<ResponseBody, TYPE> = retrofit.responseBodyConverter(typeType, EMPTY_ARRAY)
+  private val dataConverter: Converter<ResponseBody, DATA> = retrofit.responseBodyConverter(dataType, EMPTY_ARRAY)
+
+  override fun responseType(): Type = ResponseBody::class.java
+
+  override fun adapt(call: Call<ResponseBody>): EventSource<ID, TYPE, DATA> {
+    return RealEventSource(idType, typeType, dataType, idConverter, typeConverter, dataConverter, call)
+  }
+}

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

@@ -0,0 +1,124 @@
+/*
+ * Copyright (C) 2017 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package retrofit2.adapter.sse.internal
+
+import java.lang.reflect.ParameterizedType
+import java.lang.reflect.Type
+import java.lang.reflect.WildcardType
+import okhttp3.Request
+import okhttp3.ResponseBody
+import okhttp3.ResponseBody.Companion.toResponseBody
+import okhttp3.sse.EventSourceListener
+import okhttp3.sse.EventSources
+import retrofit2.Call
+import retrofit2.Converter
+import retrofit2.Response
+import retrofit2.adapter.sse.EventSource
+import retrofit2.adapter.sse.SseCallback
+
+@Suppress("NOTHING_TO_INLINE")
+private inline fun <T : Any> conversionError(value: T, type: Type): Nothing = error("Failed to convert $value to $type, actual type is ${value.javaClass}")
+
+private fun Response<ResponseBody>.asSse(listener: EventSourceListener) {
+  val okhttpResponse = raw().newBuilder().body(body() ?: error("Response body is null")).build()
+  EventSources.processResponse(okhttpResponse, listener)
+}
+
+private fun Type.acceptsString(): Boolean =
+  when (this) {
+    String::class.java -> true
+    Object::class.java -> true
+    CharSequence::class.java -> true
+    Comparable::class.java -> true
+    is ParameterizedType -> rawType === Comparable::class.java && actualTypeArguments[0].acceptsString()
+    is WildcardType -> upperBounds[0].acceptsString()
+    else -> false
+  }
+
+internal class RealEventSource<ID : Any, TYPE : Any, DATA : Any>(
+  private val idType: Type,
+  private val typeType: Type,
+  private val dataType: Type,
+  private val idConverter: Converter<ResponseBody, ID>,
+  private val typeConverter: Converter<ResponseBody, TYPE>,
+  private val dataConverter: Converter<ResponseBody, DATA>,
+  private val call: Call<ResponseBody>,
+) : EventSource<ID, TYPE, DATA> {
+  override fun request(): Request = call.request()
+
+  override fun cancel() = call.cancel()
+
+  override fun subscribe(callback: SseCallback<ID, TYPE, DATA>) {
+    call.enqueue(
+      object : retrofit2.Callback<ResponseBody> {
+        override fun onResponse(call: Call<ResponseBody>, response: Response<ResponseBody>) {
+          response.asSse(
+            object : EventSourceListener() {
+              override fun onOpen(eventSource: okhttp3.sse.EventSource, response: okhttp3.Response) {
+                callback.onOpen(this@RealEventSource)
+              }
+
+              override fun onEvent(eventSource: okhttp3.sse.EventSource, id: String?, type: String?, data: String) {
+                val convertedId = convertId(id)
+                val convertedType = convertType(type)
+                val convertedData = convertData(data)
+                callback.onEvent(this@RealEventSource, convertedId, convertedType, convertedData)
+              }
+
+              override fun onClosed(eventSource: okhttp3.sse.EventSource) {
+                callback.onClosed(this@RealEventSource)
+              }
+
+              override fun onFailure(eventSource: okhttp3.sse.EventSource, t: Throwable?, response: okhttp3.Response?) {
+                callback.onFailure(this@RealEventSource, t)
+              }
+            },
+          )
+        }
+
+        override fun onFailure(call: Call<ResponseBody>, t: Throwable) {
+          callback.onFailure(this@RealEventSource, t)
+        }
+      },
+    )
+  }
+
+  private fun convertId(id: String?): ID? {
+    @Suppress("UNCHECKED_CAST")
+    return when {
+      idType.acceptsString() -> id as ID?
+      id != null -> idConverter.convert(id.toResponseBody()) ?: conversionError(id, idType)
+      else -> null
+    }
+  }
+
+  private fun convertType(type: String?): TYPE? {
+    @Suppress("UNCHECKED_CAST")
+    return when {
+      typeType.acceptsString() -> type as TYPE?
+      type != null -> typeConverter.convert(type.toResponseBody()) ?: conversionError(type, typeType)
+      else -> null
+    }
+  }
+
+  private fun convertData(data: String): DATA {
+    @Suppress("UNCHECKED_CAST")
+    return when {
+      dataType.acceptsString() -> data as DATA
+      else -> dataConverter.convert(data.toResponseBody()) ?: conversionError(data, dataType)
+    }
+  }
+}