Using combine() operator with TestCoroutineDispatcher causes test to fail

[ttenushko]

Kotlin: 1.3.50
Coroutines: 1.3.3

Test failed when using combine operator and TestCoroutineDispatcher (test passed when using Dispatchers.Unconfined)

class FlowCombineTest {

    @Test
    fun doTest() {
        val intProvider = ObservableValue(0)
        val stringProvider = ObservableValue("")
        var data = Pair(0, "")
        val scope = CoroutineScope(TestCoroutineDispatcher() + SupervisorJob()) // test failed, use CoroutineScope(Dispatchers.Unconfined + SupervisorJob()) to pass
        scope.launch {
            combine(
                intProvider.observe(),
                stringProvider.observe()
            ) { intValue, stringValue -> Pair(intValue, stringValue) }
                .collect { pair ->
                    data = pair
                }
        }

        intProvider.set(1)
        stringProvider.set("3")
        intProvider.set(2)
        intProvider.set(3)

        scope.cancel()
        assertEquals("Test failed", Pair(3, "3"), data)
    }

    class ObservableValue<T>(initial: T) {

        var value: T = initial
            private set
        private val listeners = CopyOnWriteArraySet<Listener<T>>()

        fun set(value: T) {
            this.value = value
            listeners.forEach { it.onValueChanged(value) }
        }

        fun addListener(listener: Listener<T>) {
            listeners.add(listener)
        }

        fun removeListener(listener: Listener<T>) {
            listeners.remove(listener)
        }

        interface Listener<T> {
            fun onValueChanged(value: T)
        }
    }

    private fun <T> ObservableValue<T>.observe(): Flow<T> =
        callbackFlow {
            val listener = object : ObservableValue.Listener<T> {
                override fun onValueChanged(value: T) {
                    if (!isClosedForSend) {
                        offer(value)
                    }
                }
            }
            addListener(listener)
            listener.onValueChanged(value)
            awaitClose { removeListener(listener) }
        }
}