如何在单元测试中从 mutableSharedFlow 收集值？答案

【问题标题】：How can I collect values from a mutableSharedFlow in a unit test?如何在单元测试中从 mutableSharedFlow 收集值？
【发布时间】：2021-02-02 22:05:18
【问题描述】：

以下测试未通过。永远不会打印字符串“在添加 XXX 之前”。

    @Test
    fun testFoo() = runBlocking {
        val testCoroutineScope = TestCoroutineScope().apply {
            pauseDispatcher() // This needs to be here because the actual test handles time.
        }
        val sharedFlow = MutableSharedFlow<Int>()
        val values = mutableListOf<Int>()
        println("before launch")
        val job = testCoroutineScope.launch {
            println("before collect")
            sharedFlow.collect {
                println("before adding $it")
                values.add(it)
            }
        }
        println("before emits")
        sharedFlow.emit(1)
        sharedFlow.emit(2)
        testCoroutineScope.runCurrent()
        assertEquals(mutableListOf(1, 2), values)
        job.cancel()
    }

一旦收集部分被处理，我想有一种方法来emit 值。我不能设置replay 值，也不能使用onSubscription，因为这会改变业务逻辑。

添加 delay 或 yield 似乎没有任何区别，因此可能不是比赛条件。

如果pauseDispatcher() 被评论，则测试通过。

我的理解是，emit 将暂停，直到调用 collect lambda。

【问题讨论】：

标签： unit-testing kotlin-coroutines

【解决方案1】：

我解决它的方法是通过extraBufferCapacity 和另外一个testCoroutineScope.runCurrent()

    @Test
    fun testFoo() = runBlocking {
        val testCoroutineScope = TestCoroutineScope().apply {
            pauseDispatcher()
        }
        val sharedFlow = MutableSharedFlow<Int>(
            extraBufferCapacity = 2 // Without it, sharedFlow.emit won't have a space to save data. It will be collected
                                    // next time there's a testCoroutineScope.runCurrent()
        )
        val values = mutableListOf<Int>()
        println("before launch")
        val job = testCoroutineScope.launch {
            println("before collect")
            sharedFlow.collect {
                println("before adding $it")
                values.add(it)
            }
        }
        testCoroutineScope.runCurrent() // Allows the previous launch to start collecting

        println("before emits")
        sharedFlow.emit(1)
        sharedFlow.emit(2)
        testCoroutineScope.runCurrent()

        assertEquals(mutableListOf(1, 2), values)
        job.cancel()
    }

【讨论】：

【解决方案2】：

更新：原来我最初的尝试也不是万无一失的，竞争条件仍然存在，这是我修改后的解决方案，到目前为止一切都很好：

    val msf = MutableSharedFlow<Int>()
    runBlocking {
        val job1 = GlobalScope.launch {
            msf
                .onStart {
                    println("${Thread.currentThread().name} - on start ")
                }
                .onEach { println("got $it") }
                .collect()
        }
        val job2 = launch {
            msf.subscriptionCount
                .filter { it > 0 }
                .onEach {
                    msf.emit(42)
                    cancel()
                }
                .collect()
        }
        delay(100)
        job1.cancel()
        println(job2)
    }
    println("${Thread.currentThread().name} - escaped!")

旧解决方案（不正确）：

    runBlocking {
        val msf = MutableSharedFlow<Int>()
        val d = CompletableDeferred<Boolean>()
        launch {
            msf
                .onStart {
                    d.complete(true)
                }
                .onEach { println("got $it") }
                .collect()
        }
        d.join()
        msf.emit(42)
    }

使用 invokeOnCompletion() 处理程序不能使用挂起的函数，将它与 tryEmit() 一起使用对我不起作用（返回 true 但什么都不做）。

【讨论】：