Add Kotlin Flow adapter with SSE support

[Copilot]

Adds a new retrofit-adapters/kotlin-flow module that enables kotlinx.coroutines.flow.Flow as a suspend Retrofit service method return type, with built-in Server-Sent Events (SSE) streaming support via OkHttp okhttp-sse.

Usage

// Add to Retrofit builder
val retrofit = Retrofit.Builder()
  .baseUrl(baseUrl)
  .addCallAdapterFactory(FlowCallAdapterFactory.create())
  .build()

// SSE streaming — annotate with @Streaming, return Flow<ServerSentEvent>
interface Api {
  @Streaming
  @GET("events")
  suspend fun events(): Flow<ServerSentEvent>
}

// Non-SSE single-value flow (uses registered converter)
interface Api {
  @GET("user")
  suspend fun getUser(): Flow<String>
}

Changes

• retrofit-adapters/kotlin-flow/ – new Gradle module registered in settings.gradle
  • ServerSentEvent – data class exposing id, event, and data fields; SSE parsing and connection management are delegated to OkHttp okhttp-sse
  • FlowCallAdapterFactory – handles only suspend fun foo(): Flow<T> return types (non-suspend Flow<T> is not supported)
    • SSE mode is activated with Retrofit's @Streaming annotation; the return type must be Flow<ServerSentEvent> — a clear error is thrown at call time if a different element type is used
    • SSE flows open an OkHttp EventSource and emit parsed ServerSentEvent objects; the EventSource is cancelled when the flow is cancelled or the channel buffer overflows
    • onFailure in the SSE listener uses safe structured branching: propagates the throwable if present, wraps non-successful responses in HttpException, reports unexpected successful responses as IOException, and falls back to IllegalStateException when no info is available
    • Non-SSE flows use Retrofit's registered converter for deserialization and emit the single response body
    • A null response body on a successful response (e.g. 204) fails the flow with KotlinNullPointerException carrying the service + method name, consistent with KotlinExtensions.await() in Retrofit core
    • A lightweight FlowAsCall wrapper defers the HTTP request until the returned Flow is collected; flow cancellation correctly cancels the in-flight HTTP call
• Tests – FlowCallAdapterFactoryTest covering SSE events, multiline data, retry fields, error handling, body flows, null-body failure (asserting the exact KotlinNullPointerException message), and wrong-element-type validation

---

• CHANGELOG.md's "Unreleased" section has been updated, if applicable.

[Goooler]

@copilot I think the function should be 'suspend' only; we shouldn't support normal functions.

[Copilot]

@copilot I think the function should be 'suspend' only; we shouldn't support normal functions.

Done in ac2670d. The factory now only handles suspend fun foo(): Flow<T> — the non-suspend path (bare Flow<T> return type) has been removed. BodyFlowCallAdapter is gone, and FlowCallAdapterFactory.get() returns null for any non-Call<…> return type. Tests and docs updated accordingly.

[Copilot]

Pull request overview

This PR adds a new retrofit-adapters:kotlin-flow module which introduces a Retrofit CallAdapter.Factory to support kotlinx.coroutines.flow.Flow as the return type for Kotlin suspend service methods, including an SSE streaming mode.

Changes:

• Registers a new retrofit-adapters:kotlin-flow Gradle module (build + publishing metadata) and wires it into the build.
• Introduces FlowCallAdapterFactory with two modes: single-body Flow<T> and SSE streaming via OkHttp okhttp-sse.
• Adds initial unit tests for body flows and SSE flows, plus documentation and changelog entry.

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
settings.gradle	Includes the new :retrofit-adapters:kotlin-flow module in the build.
retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/FlowCallAdapterFactory.kt	Implements the Flow call adapter (single-body and SSE streaming).
retrofit-adapters/kotlin-flow/src/main/java/retrofit2/adapter/flow/ServerSentEvent.kt	Adds an SSE event model type exposed by the adapter.
retrofit-adapters/kotlin-flow/src/test/java/retrofit2/adapter/flow/FlowCallAdapterFactoryTest.kt	Adds unit tests for body and SSE flows using MockWebServer.
retrofit-adapters/kotlin-flow/README.md	Documents usage and behavior of the new adapter module.
retrofit-adapters/kotlin-flow/build.gradle	Adds module dependencies (Retrofit, coroutines, okhttp-sse) and jar metadata.
retrofit-adapters/kotlin-flow/gradle.properties	Provides Maven publishing coordinates and module description.
gradle/libs.versions.toml	Adds okhttp-sse to the version catalog for dependency management.
CHANGELOG.md	Notes the addition of the Kotlin Flow adapter with SSE support.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

bodyFlow enqueues a cloned call (call.clone().enqueue(...)) but the awaitClose block cancels the original call, not the cloned/enqueued instance. This means canceling the returned Flow will not cancel the in-flight HTTP request. Capture the cloned call in a local variable and cancel that in awaitClose (and use the same instance for enqueue).

Suggested change

	call
	.clone()
	.enqueue(
	object : Callback<R> {
	override fun onResponse(call: Call<R>, response: Response<R>) {
	if (!response.isSuccessful) {
	close(HttpException(response))
	return
	}
	val body = response.body()
	if (body == null) {
	close()
	return
	}
	trySend(body)
	close()
	}
	override fun onFailure(call: Call<R>, t: Throwable) {
	close(t)
	}
	}
	)
	awaitClose { call.cancel() }
	val flowCall = call.clone()
	flowCall.enqueue(
	object : Callback<R> {
	override fun onResponse(call: Call<R>, response: Response<R>) {
	if (!response.isSuccessful) {
	close(HttpException(response))
	return
	}
	val body = response.body()
	if (body == null) {
	close()
	return
	}
	trySend(body)
	close()
	}
	override fun onFailure(call: Call<R>, t: Throwable) {
	close(t)
	}
	}
	)
	awaitClose { flowCall.cancel() }

[Copilot]

When isStreaming is true, responseType is forced to ResponseBody and the adapter will produce an SSE Flow<ServerSentEvent> regardless of the declared Flow<T> element type. This can let @Streaming suspend fun foo(): Flow<Foo> compile but emit ServerSentEvent at runtime (type mismatch). Consider validating that elementType is ServerSentEvent (or otherwise constrain the streaming path) and throw a clear error if it is not.

[Copilot]

onEvent uses trySend(...) but ignores the result. If the channel buffer is full or closed, events can be dropped silently, which breaks streaming semantics. Handle the ChannelResult (e.g., close with an error on buffer overflow) or switch to a sending approach that provides backpressure without dropping (such as sending from a launched coroutine).

[Copilot]

For successful responses with a null body (e.g., 204), this flow currently completes without emitting anything. That makes suspend fun foo(): Flow<T> behave like an empty stream rather than failing like Retrofit’s non-nullable suspend fun foo(): T does. Consider failing the flow when body == null (or explicitly documenting/handling a Flow<Unit>/nullable body case) to avoid silent data loss.

[Copilot]

PR description/usage mentions an @SSE annotation and an in-module W3C-compliant SseParser, but this module’s public docs instruct using Retrofit’s @Streaming and state parsing is delegated to OkHttp okhttp-sse. Please reconcile the PR description and module documentation/implementation (either implement the promised @SSE + parser, or update the PR description to match the chosen @Streaming + OkHttp SSE approach).

[Copilot]

The PR description says ServerSentEvent follows the SSE spec with id, event, data, and retry fields, but this data class currently exposes only id, event, and data. Either add the missing retry field (and ensure it’s populated) or update the PR description/docs to match the actual API surface.

[Goooler]

@copilot apply changes based on the comments in this thread

[Copilot]

@copilot apply changes based on the comments in this thread

Applied all four actionable changes in 0db231b:

1. bodyFlow cancel bug — captured the cloned call in a flowCall variable and cancel that in awaitClose, not the original.
2. SSE type validation — get() now throws a clear IllegalStateException when @Streaming is combined with a Flow<T> where T is not ServerSentEvent.
3. Silent event drop — onEvent now checks trySend's result; on failure it cancels the EventSource and closes the channel with the cause.
4. Null body — bodyFlow now closes the flow with NullPointerException("Response body of a suspend fun was null") instead of completing normally.

Two new tests were added: bodyFlowNullBodyFails and streamingWithWrongElementTypeThrows.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 3 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

For a successful HTTP response with a null body, Retrofit's existing suspend handling throws a KotlinNullPointerException with a message that includes the service + method name (via Invocation tag). This adapter currently throws a plain NullPointerException("Response body of a suspend fun was null"), which is less actionable and inconsistent with the rest of the codebase. Consider mirroring KotlinExtensions.await()'s null-body exception/message here.

[Copilot]

Response.error(body, rawResponse) requires a non-null body and a non-successful rawResponse. If OkHttp-SSE invokes onFailure with a successful response (e.g., protocol/content-type issues) or a null body, this will throw from the callback thread instead of failing the Flow. Handle t first, then only build HttpException when response != null && !response.isSuccessful, otherwise close with an IOException/IllegalStateException describing the SSE failure.

Suggested change

	close(t ?: response?.let { HttpException(Response.error<Nothing>(it.body, it)) })
	val failure =
	when {
	t != null -> t
	response != null && !response.isSuccessful -> HttpException(response)
	response != null ->
	java.io.IOException(
	"SSE stream failed with unexpected successful response: ${response.code}",
	)
	else -> IllegalStateException("SSE stream failed without throwable or response")
	}
	close(failure)

[Copilot]

This test asserts a NullPointerException for the 204/null-body case. If the adapter is updated to mirror Retrofit's suspend behavior (throwing KotlinNullPointerException with a service/method-specific message), update the expected exception type/message here so the test reflects the intended compatibility contract.

Suggested change

	fail("Expected NullPointerException")
	} catch (_: NullPointerException) {
	// expected
	fail("Expected KotlinNullPointerException")
	} catch (e: KotlinNullPointerException) {
	assertThat(e).hasMessageThat().isEqualTo(
	"Response from retrofit2.adapter.flow.FlowCallAdapterFactoryTest\$Service.body was null but response body type was declared as non-null",
	)

[Goooler]

@copilot apply changes based on the comments in this thread

[Copilot]

@copilot apply changes based on the comments in this thread

Applied all three changes in e36eb1b:

1. onFailure safety — replaced the single-expression with a structured when block: use t if non-null, build HttpException(Response.error(response.body, response)) for non-successful responses, throw IOException for unexpected successful responses, and IllegalStateException for the no-info case. Also removed the now-unused toResponseBody import.

2. Null-body exception — bodyFlow now uses KotlinNullPointerException with the same "Response from <Service>.<method> was null but response body type was declared as non-null" message format as KotlinExtensions.await(), obtained via checkNotNull(call.request().tag(Invocation::class.java)) with a descriptive fallback message.

3. Test updated — bodyFlowNullBodyFails now catches KotlinNullPointerException and asserts the exact message using Service::class.java.name.