diff --git a/CHANGELOG.md b/CHANGELOG.md index 582b98174c..f656888708 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ - Add explicit keep rules for RxJava `Result` types to prevent their generic information from being removed. - Add `allowoptimization` flags for most kept types. - Add `Invocation.annotationUrl` which returns the original URL from the method annotation. + - Add Kotlin `Flow` adapter with SSE support. **Changed** diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 78370417f5..0cf50a0021 100644 --- a/gradle/libs.versions.toml +++ b/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" } diff --git a/retrofit-adapters/kotlin-flow/README.md b/retrofit-adapters/kotlin-flow/README.md new file mode 100644 index 0000000000..8fe4367638 --- /dev/null +++ b/retrofit-adapters/kotlin-flow/README.md @@ -0,0 +1,93 @@ +Kotlin Flow Adapter +=================== + +A `CallAdapter.Factory` for adapting [Kotlin coroutine `Flow`][1] return types in Retrofit `suspend` +service methods. + +Supported return types: + +* `Flow` — emits the single converted response body and completes. +* `Flow` (with `@Streaming`) — streams Server-Sent Events from the response body, + emitting one `ServerSentEvent` per event. + + +Usage +----- + +Add `FlowCallAdapterFactory` as a call adapter when building your `Retrofit` instance: + +```kotlin +val retrofit = Retrofit.Builder() + .baseUrl("https://example.com/") + .addCallAdapterFactory(FlowCallAdapterFactory.create()) + .build() +``` + +### Regular body flow + +Annotate a `suspend` service method with any Retrofit HTTP annotation and return `Flow`. The flow +emits the single converted response body when collected, then completes. On a non-2xx response or a +network failure the flow fails with `HttpException` or `IOException` respectively. + +```kotlin +interface MyService { + @GET("/user") + suspend fun getUser(): Flow +} +``` + +### Server-Sent Events (SSE) + +Add `@Streaming` to stream a response as [Server-Sent Events][2]. The return type must be +`Flow`. The flow emits one `ServerSentEvent` for each event dispatched by the +server and completes when the connection is closed. Cancelling the flow cancels the underlying +OkHttp `EventSource`. + +```kotlin +interface MyService { + @Streaming + @GET("/events") + suspend fun events(): Flow +} +``` + +`ServerSentEvent` exposes the fields defined by the [W3C SSE specification][2]: + +| Property | Type | Description | +| -------- | --------- | ---------------------------------------------------------- | +| `id` | `String?` | Last event ID, or `null` if not set. | +| `event` | `String?` | Event type, or `null` for the default `"message"` type. | +| `data` | `String` | Data payload; multiple `data:` lines are joined with `\n`. | + +Parsing and connection management are delegated to OkHttp's `okhttp-sse` library. The `Accept: +text/event-stream` header is added automatically. + + +Download +-------- + +Download [the latest JAR][3] or grab via [Maven][4]: + +```xml + + com.squareup.retrofit2 + adapter-kotlin-flow + latest.version + +``` + +or [Gradle][4]: + +```kotlin +implementation("com.squareup.retrofit2:adapter-kotlin-flow:latest.version") +``` + +Snapshots of the development version are available in [Sonatype's `snapshots` repository][snap]. + + + + [1]: https://kotlinlang.org/docs/flow.html + [2]: https://html.spec.whatwg.org/multipage/server-sent-events.html + [3]: https://search.maven.org/remote_content?g=com.squareup.retrofit2&a=adapter-kotlin-flow&v=LATEST + [4]: https://search.maven.org/search?q=g:com.squareup.retrofit2%20a:adapter-kotlin-flow + [snap]: https://s01.oss.sonatype.org/content/repositories/snapshots/ diff --git a/retrofit-adapters/kotlin-flow/build.gradle b/retrofit-adapters/kotlin-flow/build.gradle new file mode 100644 index 0000000000..fe6ab36213 --- /dev/null +++ b/retrofit-adapters/kotlin-flow/build.gradle @@ -0,0 +1,21 @@ +apply plugin: 'org.jetbrains.kotlin.jvm' +apply plugin: 'com.vanniktech.maven.publish' + +dependencies { + api projects.retrofit + api libs.okhttp.sse + api libs.kotlinx.coroutines + + compileOnly libs.findBugsAnnotations + + testImplementation libs.junit + testImplementation libs.truth + testImplementation libs.okhttp.mockwebserver + testImplementation libs.kotlinx.coroutines +} + +jar { + manifest { + attributes 'Automatic-Module-Name': 'retrofit2.adapter.kotlinflow' + } +} diff --git a/retrofit-adapters/kotlin-flow/gradle.properties b/retrofit-adapters/kotlin-flow/gradle.properties new file mode 100644 index 0000000000..16acc80b8c --- /dev/null +++ b/retrofit-adapters/kotlin-flow/gradle.properties @@ -0,0 +1,3 @@ +POM_ARTIFACT_ID=adapter-kotlin-flow +POM_NAME=Adapter: Kotlin Flow +POM_DESCRIPTION=A Retrofit CallAdapter for Kotlin Flow, with support for Server-Sent Events (SSE). diff --git a/retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/FlowCallAdapterFactory.kt b/retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/FlowCallAdapterFactory.kt new file mode 100644 index 0000000000..d82cf94780 --- /dev/null +++ b/retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/FlowCallAdapterFactory.kt @@ -0,0 +1,243 @@ +/* + * Copyright (C) 2026 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.flow + +import java.io.IOException +import java.lang.reflect.ParameterizedType +import java.lang.reflect.Type +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow +import okhttp3.Request +import okhttp3.ResponseBody +import okhttp3.sse.EventSource +import okhttp3.sse.EventSourceListener +import okhttp3.sse.EventSources +import okio.Timeout +import retrofit2.Call +import retrofit2.CallAdapter +import retrofit2.Callback +import retrofit2.HttpException +import retrofit2.Invocation +import retrofit2.Response +import retrofit2.Retrofit +import retrofit2.http.Streaming + +/** + * A [CallAdapter.Factory] that supports [Flow] as a `suspend` service-method return type. + * + * ## SSE (Server-Sent Events) + * + * When the method is also annotated with [@Streaming][retrofit2.http.Streaming], the adapter + * streams the HTTP response body as Server-Sent Events, emitting each parsed [ServerSentEvent] to + * the flow: + * ```kotlin + * interface Service { + * @Streaming + * @GET("events") + * suspend fun events(): Flow + * } + * ``` + * + * Register this factory with [Retrofit.Builder.addCallAdapterFactory]: + * ```kotlin + * val retrofit = Retrofit.Builder() + * .baseUrl(baseUrl) + * .addCallAdapterFactory(FlowCallAdapterFactory.create()) + * .build() + * ``` + * + * ## Non-SSE flows + * + * Without [@Streaming][retrofit2.http.Streaming], a `suspend fun foo(): Flow` return type will + * emit a single converted response body (like a regular body call) and complete, or fail with + * [HttpException] / [java.io.IOException] as appropriate. + * + * ```kotlin + * interface Service { + * @GET("user") + * suspend fun getUser(): Flow + * } + * ``` + */ +class FlowCallAdapterFactory private constructor() : CallAdapter.Factory() { + + companion object { + @JvmStatic fun create(): FlowCallAdapterFactory = FlowCallAdapterFactory() + } + + override fun get( + returnType: Type, + annotations: Array, + retrofit: Retrofit, + ): CallAdapter<*, *>? { + val isStreaming = annotations.any { it is Streaming } + + if (getRawType(returnType) != Call::class.java) return null + if (returnType !is ParameterizedType) return null + val callType = getParameterUpperBound(0, returnType) + if (getRawType(callType) != Flow::class.java) return null + if (callType !is ParameterizedType) { + error( + "Flow return type must be parameterized as Flow, Flow, or Flow" + ) + } + val elementType = getParameterUpperBound(0, callType) + if (isStreaming && getRawType(elementType) != ServerSentEvent::class.java) { + error( + "@Streaming on a Flow return type requires Flow, but found Flow<$elementType>" + ) + } + val responseType = if (isStreaming) ResponseBody::class.java else elementType + val eventSourceFactory = + if (isStreaming) EventSources.createFactory(retrofit.callFactory()) else null + return SuspendFlowCallAdapter(responseType, isStreaming, eventSourceFactory) + } +} + +private class SuspendFlowCallAdapter( + private val _responseType: Type, + private val isStreaming: Boolean, + private val eventSourceFactory: EventSource.Factory?, +) : CallAdapter>> { + + override fun responseType(): Type = _responseType + + override fun adapt(call: Call): Call> { + val flow = + if (isStreaming) { + streamingFlow(call.request(), checkNotNull(eventSourceFactory)) + } else { + bodyFlow(call) + } + return FlowAsCall(call, flow) + } +} + +/** + * A [Call] whose "response body" is a pre-built cold [Flow]. When enqueued it immediately delivers + * the flow to the callback so that Retrofit's `suspend` machinery can resume the coroutine with the + * flow value. The HTTP request is only started when the returned flow is collected. + */ +private class FlowAsCall(private val delegate: Call, private val flow: Flow<*>) : + Call> { + + override fun enqueue(callback: Callback>) { + callback.onResponse(this, Response.success(flow)) + } + + override fun execute(): Response> = Response.success(flow) + + override fun isExecuted(): Boolean = delegate.isExecuted + + override fun cancel() = delegate.cancel() + + override fun isCanceled(): Boolean = delegate.isCanceled + + override fun clone(): Call> = FlowAsCall(delegate.clone(), flow) + + override fun request(): Request = delegate.request() + + override fun timeout(): Timeout = delegate.timeout() +} + +/** + * Returns a cold [Flow] that, when collected, opens an OkHttp [EventSource] for the given [request] + * and emits each parsed [ServerSentEvent]. The connection is closed when the stream ends or the + * flow is canceled. + */ +private fun streamingFlow( + request: Request, + eventSourceFactory: EventSource.Factory, +): Flow = callbackFlow { + val eventSource = + eventSourceFactory.newEventSource( + request, + object : EventSourceListener() { + override fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String) { + val result = trySend(ServerSentEvent(id = id, event = type, data = data)) + if (result.isFailure) { + eventSource.cancel() + close(result.exceptionOrNull()) + } + } + + override fun onClosed(eventSource: EventSource) { + close() + } + + override fun onFailure( + eventSource: EventSource, + t: Throwable?, + response: okhttp3.Response?, + ) { + val failure = + when { + t != null -> t + response != null && !response.isSuccessful -> + HttpException(Response.error(response.body, response)) + response != null -> + IOException("SSE stream failed with unexpected successful response: ${response.code}") + else -> IllegalStateException("SSE stream failed without throwable or response") + } + close(failure) + } + }, + ) + awaitClose { eventSource.cancel() } +} + +/** + * Returns a cold [Flow] that, when collected, makes the HTTP call, emits the single converted + * response body, and completes. Errors result in [HttpException] or [java.io.IOException]. + */ +private fun bodyFlow(call: Call): Flow = callbackFlow { + val flowCall = call.clone() + flowCall.enqueue( + object : Callback { + override fun onResponse(call: Call, response: Response) { + if (!response.isSuccessful) { + close(HttpException(response)) + return + } + val body = response.body() + if (body == null) { + val invocation = + checkNotNull(call.request().tag(Invocation::class.java)) { + "Retrofit Invocation tag missing from request; cannot report null body location" + } + val service = invocation.service() + val method = invocation.method() + close( + KotlinNullPointerException( + "Response from ${service.name}.${method.name}" + + " was null but response body type was declared as non-null" + ) + ) + return + } + trySend(body) + close() + } + + override fun onFailure(call: Call, t: Throwable) { + close(t) + } + } + ) + + awaitClose { flowCall.cancel() } +} diff --git a/retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/ServerSentEvent.kt b/retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/ServerSentEvent.kt new file mode 100644 index 0000000000..10452a06fb --- /dev/null +++ b/retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/ServerSentEvent.kt @@ -0,0 +1,26 @@ +/* + * Copyright (C) 2026 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.flow + +/** + * Represents a single Server-Sent Event as defined by the + * [W3C SSE specification](https://html.spec.whatwg.org/multipage/server-sent-events.html). + * + * @property id The last event ID, or `null` if none was set. + * @property event The event type, or `null` if the default "message" type. + * @property data The data payload. Multiple `data:` lines are joined with `\n`. + */ +data class ServerSentEvent(val id: String?, val event: String?, val data: String) diff --git a/retrofit-adapters/kotlin-flow/src/test/java/retrofit2/adapter/flow/FlowCallAdapterFactoryTest.kt b/retrofit-adapters/kotlin-flow/src/test/java/retrofit2/adapter/flow/FlowCallAdapterFactoryTest.kt new file mode 100644 index 0000000000..5613dbd26c --- /dev/null +++ b/retrofit-adapters/kotlin-flow/src/test/java/retrofit2/adapter/flow/FlowCallAdapterFactoryTest.kt @@ -0,0 +1,204 @@ +/* + * Copyright (C) 2026 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.flow + +import com.google.common.truth.Truth.assertThat +import java.io.IOException +import java.lang.reflect.Type +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import okhttp3.ResponseBody +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import okhttp3.mockwebserver.SocketPolicy +import org.junit.Assert.fail +import org.junit.Rule +import org.junit.Test +import retrofit2.Converter +import retrofit2.HttpException +import retrofit2.Retrofit +import retrofit2.http.GET +import retrofit2.http.Streaming + +class FlowCallAdapterFactoryTest { + @get:Rule val server = MockWebServer() + + interface Service { + @Streaming @GET("/") suspend fun sseEvents(): Flow + + @GET("/") suspend fun body(): Flow + } + + /** Service method that incorrectly uses @Streaming with a non-ServerSentEvent element type. */ + interface BadStreamingService { + @Streaming @GET("/") suspend fun events(): Flow + } + + private val retrofit + get() = + Retrofit.Builder() + .baseUrl(server.url("/")) + .addConverterFactory(StringConverterFactory()) + .addCallAdapterFactory(FlowCallAdapterFactory.create()) + .build() + + @Test + fun sseEvents() = runBlocking { + server.enqueue( + MockResponse() + .setHeader("Content-Type", "text/event-stream") + .setBody("id: 1\nevent: ping\ndata: hello\n\n" + "id: 2\ndata: world\n\n") + ) + val service = retrofit.create(Service::class.java) + val events = service.sseEvents().toList() + assertThat(events) + .containsExactly( + ServerSentEvent(id = "1", event = "ping", data = "hello"), + ServerSentEvent(id = "2", event = null, data = "world"), + ) + .inOrder() + } + + @Test + fun sseEventsMultilineData() = runBlocking { + server.enqueue( + MockResponse() + .setHeader("Content-Type", "text/event-stream") + .setBody("data: line one\ndata: line two\n\n") + ) + val service = retrofit.create(Service::class.java) + val events = service.sseEvents().toList() + assertThat(events) + .containsExactly(ServerSentEvent(id = null, event = null, data = "line one\nline two")) + Unit + } + + @Test + fun sseEventsWithRetry() = runBlocking { + server.enqueue( + MockResponse() + .setHeader("Content-Type", "text/event-stream") + .setBody("retry: 3000\ndata: reconnect\n\n") + ) + val service = retrofit.create(Service::class.java) + val events = service.sseEvents().toList() + assertThat(events).containsExactly(ServerSentEvent(id = null, event = null, data = "reconnect")) + Unit + } + + @Test + fun sseEventsCommentsIgnored() = runBlocking { + server.enqueue( + MockResponse() + .setHeader("Content-Type", "text/event-stream") + .setBody(": this is a comment\ndata: real\n\n") + ) + val service = retrofit.create(Service::class.java) + val events = service.sseEvents().toList() + assertThat(events).containsExactly(ServerSentEvent(id = null, event = null, data = "real")) + Unit + } + + @Test + fun sseEventsHttpError() = runBlocking { + server.enqueue(MockResponse().setResponseCode(500)) + val service = retrofit.create(Service::class.java) + try { + service.sseEvents().toList() + fail("Expected HttpException") + } catch (e: HttpException) { + assertThat(e.code()).isEqualTo(500) + } + } + + @Test + fun sseEventsNetworkFailure() = runBlocking { + server.enqueue(MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST)) + val service = retrofit.create(Service::class.java) + try { + service.sseEvents().toList() + fail("Expected IOException") + } catch (_: IOException) { + // expected + } + } + + @Test + fun bodyFlow() = runBlocking { + server.enqueue(MockResponse().setBody("hello")) + val service = retrofit.create(Service::class.java) + val values = service.body().toList() + assertThat(values).containsExactly("hello") + Unit + } + + @Test + fun bodyFlowHttpError() = runBlocking { + server.enqueue(MockResponse().setResponseCode(404)) + val service = retrofit.create(Service::class.java) + try { + service.body().toList() + fail("Expected HttpException") + } catch (e: HttpException) { + assertThat(e.code()).isEqualTo(404) + } + } + + @Test + fun bodyFlowNullBodyFails() = runBlocking { + // 204 No Content — response body is null + server.enqueue(MockResponse().setResponseCode(204)) + val service = retrofit.create(Service::class.java) + try { + service.body().toList() + fail("Expected KotlinNullPointerException") + } catch (e: KotlinNullPointerException) { + assertThat(e) + .hasMessageThat() + .isEqualTo( + "Response from ${Service::class.java.name}.body" + + " was null but response body type was declared as non-null" + ) + } + } + + @Test + fun streamingWithWrongElementTypeThrows() = runBlocking { + val service = retrofit.create(BadStreamingService::class.java) + try { + service.events() + fail("Expected IllegalArgumentException for @Streaming with non-ServerSentEvent element type") + } catch (e: IllegalArgumentException) { + // Retrofit wraps the IllegalStateException from our factory in IllegalArgumentException. + assertThat(e).hasCauseThat().hasMessageThat().contains("Flow") + } + } + + private class StringConverterFactory : Converter.Factory() { + override fun responseBodyConverter( + type: Type, + annotations: Array, + retrofit: Retrofit, + ): Converter? { + return if (type == String::class.java) { + Converter { it.string() } + } else { + null + } + } + } +} diff --git a/settings.gradle b/settings.gradle index e94afa5343..5bec9958b9 100644 --- a/settings.gradle +++ b/settings.gradle @@ -26,6 +26,7 @@ include ':retrofit-response-type-keeper' include ':retrofit-adapters:guava' include ':retrofit-adapters:java8' +include ':retrofit-adapters:kotlin-flow' include ':retrofit-adapters:rxjava' include ':retrofit-adapters:rxjava2' include ':retrofit-adapters:rxjava3'