Is converting a `Flow` to an `Observable` safe? The reason I’m wondering is because the `Observable`...

# coroutines

[spierce7]

Is converting a Flow to an Observable safe? The reason I’m wondering is because the Observable can be cancelled, but the Flow can’t. Is the suspending behavior of the coroutine sufficient?

With Kotlin Multiplatform for the JS side we’re converting `Flow`s to rxjs Observable s. Code inside the thread.

[spierce7]

fun <T> Flow<T>.toObservable(): Observable<T> {
    val flow = this

    return Observable { _, subscriber ->
        GlobalScope.launch(start = CoroutineStart.UNDISPATCHED) {
            try {
                flow.collect {
                    if (!subscriber.closed) {
                        subscriber.next(it)
                    }
                }

                if (!subscriber.closed) {
                    subscriber.complete()
                }
            } catch (e: Throwable) {
                if (!subscriber.closed) {
                    subscriber.error(e)
                }
            }
        }
    }
}

[louiscad]

A Flow can definitely be cancelled by cancelling the collector directly or indirectly (e.g. through an operator). Your snippet doesn't seem to link to Observable cancellation though.

[spierce7]

oh wait - I should have a scope created inside the Observable creation scope, and cancel the internal scope, shouldn’t I?

[Zach Klippenstein (he/him) [MOD]]

Have you looked at the coroutines libraries implementation of this method for RxJava?

[spierce7]

@Zach Klippenstein (he/him) [MOD] I looked for it, but couldn’t find it.

[spierce7]

@louiscad What do you think?

return Observable { _, subscriber ->
        val job = Job()
        GlobalScope.launch(job, start = CoroutineStart.UNDISPATCHED) {
            try {
                flow.collect {
                    if (!subscriber.closed) {
                        subscriber.next(it)
                    }
                }

                if (!subscriber.closed) {
                    subscriber.complete()
                }
            } catch (e: Throwable) {
                if (!subscriber.closed) {
                    subscriber.error(e)
                }
            }
        }

        subscriber.add {
            job.cancel()
        }
    }

[Zach Klippenstein (he/him) [MOD]]

https://github.com/Kotlin/kotlinx.coroutines/blob/0cab30bc79575dfe7702766a99a7b9753218bb13/reactive/kotlinx-coroutines-rx3/src/RxConvert.kt#L104

[louiscad]

@spierce7 I don't know the Observable API, but I think you can write unit tests to check it works as expected for different cases like cancellation, exceptions, and you can take inspiration from what Zach just linked that is definitely tested.

[elizarov]

What do you mean by "`Observable` can be cancelled"?