ChildCancelledException when using Rx Observable.asFlow() extension within combine() #2104
Comments
|
As a mitigation, I've switched from using |
|
We ran into this as well... if I understand the problem correctly, it's more general than just I have created a custom import com.example.rx2.asFlow // via Publisher
import kotlinx.coroutines.rx2.asFlow as _asFlow
class ToFlowConversionsTest {
private val uncaughtRxExceptions = CopyOnWriteArrayList<Throwable>()
// Purposefully do not customize any schedulers - the tests will use Thread.sleep.
// I can't seem to reproduce the bug any other way.
@[JvmField Rule] val rxJavaPluginsRule = RxJavaPluginsRule(
errorHandler = { uncaughtRxExceptions += it }
)
@[JvmField Rule] val coroutineRule = TestCoroutineRule {
TestProvidedCoroutineScope(context = TestCoroutineExceptionHandler())
}
@Test
fun testThatWorkaroundFixesTheProblem() {
doTest { asFlow() }
assertEquals<List<Throwable>>(emptyList(), uncaughtRxExceptions)
}
@Test
fun testThatBuiltInObservableAsFlowIsStillBroken() {
doTest { _asFlow() }
val failureMessage = uncaughtRxExceptions.toString()
// Sometimes there can be more than one!
assertNotEquals(0, uncaughtRxExceptions.size, failureMessage)
val expectedPrefix = "kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled"
for (thrown in uncaughtRxExceptions) {
assertTrue(thrown.toString().startsWith(expectedPrefix), failureMessage)
}
}
private fun doTest(convert: Observable<Long>.() -> Flow<Long>) {
val observable = Observable.interval(1L, TimeUnit.MILLISECONDS)
val flow = observable.convert()
val items = mutableListOf<Long>()
val job = flow
.onEach { items += it }
.launchIn(coroutineRule)
Thread.sleep(20L)
coroutineRule.advanceUntilIdle()
job.cancel()
}
}Although I was initially getting pretty good results from Regardless, it seems like none of these exceptions should be propagating to Rx's global error handling. |
|
Seeing another crash with a slightly different stack trace that seems related: Unfortunately, given the depth of this stack trace, it is difficult to trace it down to the root source. |
|
Here is another one: This concerns me because RxJava 2 is used heavily in our code base and recently we have encouraged developers to prefer Coroutines instead. However the Rx2 <-> Coroutines interop is a must in order for us to move forward with Coroutines for new code. |
|
On a similar trace, turning on Caused by: kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
(Coroutine boundary)
at kotlinx.coroutines.channels.AbstractSendChannel.offer(AbstractChannel.kt:156)
at kotlinx.coroutines.channels.ChannelCoroutine.offer(Unknown Source:2)
at kotlinx.coroutines.channels.ChannelsKt__ChannelsKt.sendBlocking(Channels.kt:21)
at kotlinx.coroutines.channels.ChannelsKt.sendBlocking(Unknown Source:1)
at kotlinx.coroutines.rx2.RxConvertKt$asFlow$1$observer$1.onNext(RxConvert.kt:98)Where result is Closed<*> -> throw recoverStackTrace(helpCloseAndGetSendException(result))This might suggest that the underlying issue is actually about |
|
The documentation in kotlinx.coroutines/kotlinx-coroutines-core/common/src/flow/Builders.kt Lines 302 to 310 in 1b34e1c Looks like the implementation of kotlinx.coroutines/reactive/kotlinx-coroutines-rx2/src/RxConvert.kt Lines 78 to 90 in 1b34e1c Within the The reason
|
I apologize but I do not have a reduced repro for this at the moment. I am encountering a race condition that occurs in an Android app I am working on where we are slowly migrating from RxJava 2 to Coroutines. Essentially we have an upstream
MutableStateFlowthat wedebounceand then combine with anotherMutableStateFlowand emit aPair. From there, we applyflatMapLatestwhich involves a downstreamFlowthat is acombineagainst severalFlowswhere some of theseFlowsare Rx Observables that are converted toFlowsusing theObservable.asFlow()extension, which btw, is still experimental.The logic/pseudocode is as follows:
Here is the stack trace I am seeing:
The text was updated successfully, but these errors were encountered: