Scoped flow leaks the decomposition into behaviour

[qwwdfsad]

Consider the following operators:

fun <T> Flow<T>.id(): Flow<T> = flow { collect { emit(it) } }

fun <T> Flow<T>.idScoped(): Flow<T> = flow {
    coroutineScope {
        val channel = Channel<T>()
        launch {
            try {
                for (i in channel) emit(i)
            } finally {
                channel.cancel()
            }
        }

        val result = runCatching {
            collect {
                channel.send(it)
            }
        }
        channel.close(result.exceptionOrNull())
        result.getOrThrow()
    }
}

Our goal, eventually, is to make them more or less indistinguishable (modulo dispatch order, of course).

Currently, it is not the case, this behavior leaks directly into user code:

flow {
    emit(1)
    delay(Long.MAX_VALUE)
}.id[Scoped]().collect { value ->
    kotlin.coroutines.coroutineContext.cancel()
}

Depending on the chosen operator, behaviour will be different (CE or hang).
To fix it, invokeOnCompletion { if (it is CancellationException) cancel() } should be used right after launch in a scoped variant.

While this can problem can be workarounded, it clearly indicates that we are missing an important primitive or cancellation mode.

[elizarov]

It seems that we already have corresponding primitives:

• produce on one side a channel
• consumeEach on the other side of a channel

A correct implementation of idScoped seems to be the following:

fun <T> Flow<T>.idScoped(): Flow<T> = flow {
    coroutineScope {
        val channel = produce<T> {
            collect {
                send(it)
            }
        }
        channel.consumeEach { 
            emit(it)
        }
    }
}

[qwwdfsad]

I am not sure it is correct (I deliberately avoided using produce here).

With produce the problem just hides a bit better, e.g.

flow {
    emit(1)
    kotlin.coroutines.coroutineContext.cancel()
}.idScoped().collect { value ->
    println(value)
}

still behaves differently depending on the implementation of idScoped

[elizarov]

The produce/consumeEach pair is supposed to work correctly in this case by design. If it does not work correctly, then it shall be fixed. In fact, it is the other side of the problem in #957 about confusing ClosedSendChannelException. I've committed fix in PR #1158 and added test to ensure that produce/consumeEach pair is transparent to cancellations.

[elizarov]

Here is a larger design problem that we have here. produce is currently an "atomic" primitive of kotlinx.coroutines in the sense that you cannot implement it via public API of launch and Channel() factory. We miss some lower-level "cancellation control" API to make it possible. However, the design of such a lower-level API is out of the scope now. We can simply use produce/consumeEach as needed in flow operator implementations.