Hey everyone :wave: I have a question regarding `K...

# coroutines

[Joaquim Ley]

Hey everyone 👋 I have a question regarding `Kotlin.Flow`s
I’m moving a data stream through the layers of an app, and each layer does its thing transforming data whatnot. My data source(s) return a flow, my problem here is:
- If I do the mapping before declaring something asFlow() (currently converting a Rx Flowable to a Flow) everything works, but when it reaches the second layer I get an exception :

kotlin.coroutines.intrinsics.CoroutineSingletons cannot be cast to MY_MODEL

- The way I apply these transformations is with the .map() operator

To make it more clear

Layer 1 - DB -> Flowable -> .map() -> asFlow() - ✅

Layer 2 - flow (from layer 1) -> .map() -> flow() - 💥

I’ve been searching google and GitHub but it seems like this hasn’t been solved.

logs

Caused by: kotlinx.coroutines.JobCancellationException: Parent job is Cancelling; job=DispatchedCoroutine{Cancelled}@173d7e3
     Caused by: java.lang.ClassCastException: kotlin.coroutines.intrinsics.CoroutineSingletons cannot be cast to MODEL_CLASS_HERE_
        at CONTACT_REPOSITORY_CLASSl$fetchFlowContacts$1$1$invokeSuspend$$inlined$collect$1.emit(Collect.kt:73)
        at kotlinx.coroutines.flow.FlowKt__TransformKt$map$$inlined$unsafeTransform$1$2.emit(Collect.kt:98)
        at kotlinx.coroutines.reactive.flow.PublisherAsFlow.collect(PublisherAsFlow.kt:78)
        at kotlinx.coroutines.reactive.flow.PublisherAsFlow$collect$1.invokeSuspend(Unknown Source:12)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:241)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:740)

-------------

A workaround is using the flatMapConcat() instead of map() and return a flowOf(someTransformation()) which feels a little dirty creating a new flow every-time data is emitted.

---------------
// Kotlin
versions.kotlin = “1.3.41”
versions.kotlin_coroutines = “1.2.1"
versions.kotlin_coroutines_reactive = “1.3.0-RC”
versions.kotlin_rx = “2.1.0"

Ran on Android

[Joaquim Ley]

Ok not sure how much better this is but

flow.transform { emit( /* transformation here */) }

works.

[tseisel]

Why are you using different versions for kotlinx-coroutines-core and kotlinx-coroutines-reactive ?

[elizarov]

^ This is the reason. Their versions should match

[Joaquim Ley]

Thank you @tseisel @elizarov — I’m on a spike branch and didn’t thought of this detail, also sorry for afking, going to test now will get back to you 🙂

[Joaquim Ley]

yay it works! 🎉