kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled #1433
Comments
|
Apparently, it is a bug in the underlying implementation mechanism of |
|
@qwwdfsad I see a random return channel.consumeAsFlow()
.first { capture ->
if (this(capture)) {
true
} else {
capture.close()
false
}
}The following code (using the channel directly) gives me another stack trace: return channel
.first { capture ->
if (this(capture)) {
true
} else {
capture.close()
false
}
}Is this related, or should I create a new ticket (and even try to create a sample for it)? |
|
@sellmair It's better to open a new issue with a reproducer I can run |
|
I still see this on 1.3.2. I'm not sure if I'm doing something wrong, but the use case is very similar to the one described. The issue happens when the callback inside a Code: private val geoQueryData: Flow<HashMap<Key, GeoLocation>> = geoQueryChannel.asFlow().flatMapLatest {
channelFlow {
val searchRadiusChannel = searchRadiusInKms.openSubscription()
val searchRadiusObserver = launch {
for (searchRadius in searchRadiusChannel) {
delay(1000)
it.radius = searchRadius
}
}
val map = hashMapOf<Key, GeoLocation>()
val listener = object : GeoQueryEventListener {
override fun onKeyEntered(key: String, location: GeoLocation) {
map[key] = location
}
override fun onKeyExited(key: String) {
map.remove(key)
}
override fun onKeyMoved(key: String, location: GeoLocation) {
map[key] = location
}
override fun onGeoQueryReady() {
offer(map)
}
override fun onGeoQueryError(error: DatabaseError) {
cancel(CancellationException("API Error", error.toException()))
}
}
it.addGeoQueryEventListener(listener)
awaitClose {
searchRadiusObserver.cancel()
searchRadiusChannel.cancel()
it.removeGeoQueryEventListener(listener)
}
}
}.conflate().flowOn(Dispatchers.IO)
fun DatabaseReference.asChannelFlow(geoLocation: GeoLocation) = channelFlow {
val listener = this@asChannelFlow.addValueEventListener(
object : ValueEventListener {
override fun onDataChange(userSnap: DataSnapshot) {
logd("Got snap $userSnap")
userSnap.getValue(User::class.java)?.apply {
pos = geoLocation.asLatLng()
}?.let {
logd("Got user, offering $it")
offer(it)
}
}
override fun onCancelled(p0: DatabaseError) {
cancel("OnCancelled $p0", p0.toException())
}
}
)
logd("Got - asChannelFlow")
awaitClose {
logd("Got - asChannelFlow closing")
this@asChannelFlow.removeEventListener(listener)
}
}
val nearbyUsers: Flow<Resource<List<User>>> = geoQueryData.flatMapLatest {
logd("Got from GeoQuery $it")
combine(
it.map { (key, geoLocation) -> db.userRef(key).asChannelFlow(geoLocation) }
) { usersArray ->
logd("Got usersArray $usersArray")
Resource.Success(usersArray.asList().sortedByDistance(centerLocation))
}
}.conflate().flowOn(Dispatchers.IO)Stacktrace shows the line that caused the crash was:
Inside the Firebase listener, inside the Reading #1454, It seems perhaps it is as expected. Using |
|
|
Hi, in my app I randomly get a
ChildCancellationException. It comes from theflatMapLatestoperator (previously namedswitchMap). I tested it on the JVM with versions 1.3.0-M2, 1.3.0-RC, 1.3.0-RC2. I was able to reproduce it in a test:Of course this test is not my real code, but this appears to reproduce the exception (almost) every time it is run. You may need to run the test again if you do not get the exception. It should happen within a few seconds of running the test.
Is this expected behaviour and/or is there anything I can do to fix this?
There was a discussion about this on Slack with @qwwdfsad: https://kotlinlang.slack.com/archives/C1CFAFJSK/p1563388954310700
The text was updated successfully, but these errors were encountered: