Repository: ktorio/ktor
File: ktor-client/ktor-client-okhttp/jvm/src/io/ktor/client/engine/okhttp/OkHttpSSESession.kt (Kotlin)

Code before:
import kotlin.coroutines.CoroutineContext

internal class OkHttpSSESession private constructor(
    factory: EventSource.Factory,
    engineRequest: Request,
    override val coroutineContext: CoroutineContext,
) : SSESession, EventSourceListener() {

    constructor(
        engine: OkHttpClient,
        engineRequest: Request,
        callContext: CoroutineContext,
    ) : this(
        factory = EventSources.createFactory(engine),
        engineRequest = engineRequest,
        coroutineContext = callContext + Job() + CoroutineName("OkHttpSSESession"),
    )

    private val serverSentEventsSource = factory.newEventSource(engineRequest, this)

    internal val originResponse: CompletableDeferred<Response> = CompletableDeferred()

    private val _incoming = Channel<ServerSentEvent>(8)

    override val incoming: Flow<ServerSentEvent> = _incoming.consumeAsFlow()
        .onCompletion { cause ->
            // Use onCompletion operator to handle CancellationExceptions which occur in downstream flow.
            if (cause is CancellationException) close()
        }

    init {
        coroutineContext.job.invokeOnCompletion {
            close()
        }
    }

    override fun onOpen(eventSource: EventSource, response: Response) {
        originResponse.complete(response)
    }

    override fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String) {
        _incoming.trySendBlocking(ServerSentEvent(data, type, id))
            .onFailure { if (it is CancellationException) throw it }
    }

    override fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) {
        val statusCode = response?.code
        val contentType = response?.headers?.get(HttpHeaders.ContentType)

        if (response != null &&
            (statusCode != HttpStatusCode.OK.value || contentType != ContentType.Text.EventStream.toString())

Review comment (suggestion):
Shouldn't we do a little more complex check, like in `DefaultClientSSESession`, so it will be:
```kotlin
override val incoming: Flow<ServerSentEvent> = _incoming.consumeAsFlow()
        .catch { cause ->
            if (cause !is CancellationException) {
                close()
                throw cause
            }
        }.onCompletion { cause ->
            // Use onCompletion operator to handle CancellationExceptions which occur in downstream flow.
            if (cause is CancellationException) {
                close()
            }
        }
```

Code after:
import kotlin.coroutines.CoroutineContext

internal class OkHttpSSESession private constructor(
    factory: EventSource.Factory,
    engineRequest: Request,
    override val coroutineContext: CoroutineContext,
) : SSESession, EventSourceListener() {

    constructor(
        engine: OkHttpClient,
        engineRequest: Request,
        callContext: CoroutineContext,
    ) : this(
        factory = EventSources.createFactory(engine),
        engineRequest = engineRequest,
        coroutineContext = callContext + Job() + CoroutineName("OkHttpSSESession"),
    )

    private val serverSentEventsSource = factory.newEventSource(engineRequest, this)

    internal val originResponse: CompletableDeferred<Response> = CompletableDeferred()

    private val _incoming = Channel<ServerSentEvent>(8)

    override val incoming: Flow<ServerSentEvent> = _incoming.consumeAsFlow()
        .onCompletion { cause ->
            // Use onCompletion operator to handle CancellationExceptions which occur in downstream flow.
            if (cause is CancellationException) close(cause = null)
        }

    init {
        coroutineContext.job.invokeOnCompletion {
            close(cause = null)
        }
    }

    override fun onOpen(eventSource: EventSource, response: Response) {
        originResponse.complete(response)
    }

    override fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String) {
        _incoming.trySendBlocking(ServerSentEvent(data, type, id))
            .onFailure { if (it is CancellationException) throw it }
    }

    override fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) {
        val statusCode = response?.code
        val contentType = response?.headers?.get(HttpHeaders.ContentType)

        if (response != null &&
            (statusCode != HttpStatusCode.OK.value || contentType != ContentType.Text.EventStream.toString())