如何将可观察的排放传递给MutableSharedFlow？

Question

好吧，我有一个可观察的，我已经使用asflow（）将其转换但不发出。 我正在尝试从RX和频道迁移到Flow，因此我有此功能

override fun processIntents(intents: Observable<Intent>) {
     intents.asFlow().shareTo(intentsFlow).launchIn(this)
}

shareto（）是一个扩展功能，它oneach {recever.emit（it）} ，processintents存在于基本视图模型中，IntentsFlow是mutableSharedFlow。

fun <T> Flow<T>.shareTo(receiver: MutableSharedFlow<T>): Flow<T> {
    return onEach { receiver.emit(it) }
}

我想传递来自Intents可观察到IntentsFlow的排放，但它根本不起作用，并且单位测试持续失败。

@Test(timeout = 4000)
fun `WHEN processIntent() with Rx subject or Observable emissions THEN intentsFlow should receive them`() {
    return runBlocking {

        val actual = mutableListOf<TestNumbersIntent>()
        val intentSubject = PublishSubject.create<TestNumbersIntent>()


        val viewModel = FlowViewModel<TestNumbersIntent, TestNumbersViewState>(
            dispatcher = Dispatchers.Unconfined,
            initialViewState = TestNumbersViewState()
        )

        viewModel.processIntents(intentSubject)


        intentSubject.onNext(OneIntent)
        intentSubject.onNext(TwoIntent)
        intentSubject.onNext(ThreeIntent)

        viewModel.intentsFlow.take(3).toList(actual)

        assertEquals(3, actual.size)
        assertEquals(OneIntent, actual[0])
        assertEquals(TwoIntent, actual[1])
        assertEquals(ThreeIntent, actual[2])
    }
}

测试在4000毫秒之后定时出现 org.junit.runners.model.testtimedoutexception：测试时间 4000毫秒

Answer 1

这项工作

val ps = PublishSubject.create<Int>()
val mf = MutableSharedFlow<Int>()
val pf = ps.asFlow()
    .onEach {
        mf.emit(it)
    }
launch {
    pf.take(3).collect()
}
launch {
    mf.take(3).collect {
        println("$it") // Prints 1 2 3
    }
}
launch {
    yield() // Without this we suspend indefinitely
    ps.onNext(1)
    ps.onNext(2)
    ps.onNext(3)
}

我们需要以（3） s来确保我们的程序终止，因为mutableSharedFlow and publishsubject-ject -＆gt;流无限期收集。

我们需要收益，因为我们正在使用一个线程，我们需要给其他Coroutines一个开始工作的机会。

---

取2

这是更好的。不使用取，并自行清理。

发出最后一项后，在publishsubject terminates mutableSharedFlow集合上调用oncomplete。这是一个便利，因此当此代码运行时，它完全终止。这不是要求。您可以根据需要安排工作终止。

您的代码永不终止与mutableSharedFlow从未收集的排放无关。这些是单独的问题。首先是由于以下事实：既不是publishsubject的流量，也不是mutableSharedFlow自行终止。 PublishSubject流程将在onComplete时终止。 mutableSharedFlow将终止当Coroutine（特别是其job）终止它。

Flow由publishsubject.asflow（）构建，等待排放。这引入了竞赛条件，即准备收集和调用代码PublishSubject.onnext（）。

我相信，这就是为什么流程收集未在代码中拾取Onnext排放的原因。

这就是为什么在我们启动从psf收集的Coroutine之后，需要立即需要A 收益。

val ps = PublishSubject.create<Int>()
val msf = MutableSharedFlow<Int>()
val psf = ps.asFlow()
    .onEach {
        msf.emit(it)
    }
val j1 = launch {
    psf.collect()
}

yield() // Use this to allow psf.collect to catch up

val j2 = launch {
    msf.collect {
        println("$it") // Prints 1 2 3 4
    }
}

launch {
    ps.onNext(1)
    ps.onNext(2)
    ps.onNext(3)
    ps.onNext(4)
    ps.onComplete()
}
j1.invokeOnCompletion { j2.cancel() }
j2.join()