KTOR-8244: Fixed an issue where canceling the incoming flow of OkHttpSSESession/DefaultClientSSESession failed to disconnect the SSE session.

[hehua2008]

Fixed an issue where canceling the incoming flow of OkHttpSSESession failed to disconnect the SSE session.

Subsystem
Client
ktor-client/ktor-client-okhttp

Motivation
Refer to a bug/ticket #4692.
In the OkHttpSSESession implementation of SSESession, cancel incoming flow will NOT trigger serverSentEventsSource.cancel(), which will keep SSE still connecting !

Solution

• First, val incoming: Flow<ServerSentEvent> in OkHttpSSESession should always return the same Flow instance, just as DefaultClientSSESession did.

• Second, cancel _incoming.receiveAsFlow() will just cancel the Flow but will NOT cancel the _incoming Channel, so we should use _incoming.consumeAsFlow() instead.

• Third, if incoming.trySendBlocking(ServerSentEvent(data, type, id)) fails with a CancellationException, we should rethrow the CancellationException to let fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) handle it and close the SSE session.

[e5l]

Hey @hehua2008, lgtm. Could you add a test, so we can merge? :)

[hehua2008]

For same bug in DefaultClientSSESession:
Because catch operator only catch throwable occurs in upstream flow, so we use onCompletion operator instead to handle CancellationException occurs in either upstream flow or downstream flow.

    }.catch { cause ->
        when (cause) {
            is CancellationException -> {
                // CancellationException will be handled by onCompletion operator
            }

            else -> {
                LOGGER.trace { "Error during SSE session processing: $cause" }
                close()
                throw cause
            }
        }
    }.onCompletion { cause ->
        // Because catch operator only catch throwable occurs in upstream flow, so we use onCompletion operator instead
        // to handle CancellationException occurs in either upstream flow or downstream flow.
        if (cause is CancellationException) {
            close()
        }
    }

[hehua2008]

Hey @hehua2008, lgtm. Could you add a test, so we can merge? :)

I have added testCancelSseSession in ServerSentEventsTest for ktor-client:

    @Test
    fun testCancelSseSession() = clientTests {
        config {
            install(SSE)
        }

        test { client ->
            coroutineScope {
                val job = launch {
                    val session = client.serverSentEventsSession("$TEST_SERVER/sse/hello?times=20&interval=100")
                    try {
                        session.incoming.collect()
                    } finally {
                        withContext(NonCancellable) {
                            delay(500)
                            assertFalse(session.isActive)
                        }
                    }
                }
                delay(500)
                job.cancelAndJoin()
            }
        }
    }

[e5l]

lgtm

[osipxd]

Hey @hehua2008, thank you for the PR! Could you please fix import ordering in DefaultClientSSESession?

[hehua2008]

Hey @hehua2008, thank you for the PR! Could you please fix import ordering in DefaultClientSSESession?

Sorry for that, I have fixed it.

[Savrov]

When we can expect it to be released?

[Stexxe]

In the next patch release of Ktor 3.1.2.

[Savrov]

I would like to know some est. date, cause it's a critical defect in our app and we can not release it without that fix.

Thank you

[e5l]

Hey, @Savrov! We usually do monthly releases, and the next one is planned for the last week of March.
If you want to check out the build with a fix, please consider checking https://ktor.io/eap. They are the same builds from the main branch, but published daily.