shareIn keeps coroutines alive even when there are no subscribers

[dkhalanskyjb]

Consider this code:

        runBlocking {
            launch(CoroutineName("coroutines in flow")) {
                val flow = flow {
                    var i = 0
                    while (true) {
                        emit(i)
                        delay(100.milliseconds)
                        ++i
                    }
                }.shareIn(this, SharingStarted.WhileSubscribed(stopTimeout = 1.seconds))
                flow.first()
            }
            delay(2.seconds)
            println("breakpoint here")
        }

If we put a breakpoint at the designated println, in the debugger, we see several coroutines with coroutines in flow in their name. This is surprising: this is almost two seconds after the only subscriber of the flow in existence has finished its work, and shareIn was parameterized only to keep sharing for one second after the last subscriber is lost. However, the flow still keeps some coroutines.

Kudos to @zsmb13 for reporting this!

[qwwdfsad]

A small historical note.

During our initial design, WhileSubscribed was designed for very specific UI-related use cases where SharedFlow lifecycle is tightly bound to the corresponding UI element Job, meaning that the liveness of the helper coroutines was never considered as a dedicated design choice. It is rather how things are played out (i.e. it was quite natural to express WhileSubscribed in terms of operators over infinite subscription flow).

We can (if we decide on it) either fix it ad-hoc or to use it as a driving force for #1065