# Interoperation between RxJava and Kotlin Coroutines

DevFeed: [Interoperation between RxJava and Kotlin Coroutines](<https://devfeed.tech/articles/interoperation-between-rxjava-and-kotlin-coroutines-24806.md>)

Original publisher: [Read original article](<https://akarnokd.blogspot.com/2017/09/interoperation-between-rxjava-and.html>)

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-11T21:29:00Z

Content type: tutorial

Language: en

Sources: [Akarnokd - Advanced RxJava](<https://devfeed.tech/sources/akarnokd-advanced-rxjava.md>)

Topics: [kotlin-coroutines](<https://devfeed.tech/topics/kotlin-coroutines.md>), [Coroutines](<https://devfeed.tech/topics/coroutines.md>), [Kotlin](<https://devfeed.tech/topics/kotlin.md>), [Code](<https://devfeed.tech/topics/code.md>), [Compiler](<https://devfeed.tech/topics/compiler.md>)

Tags: [asynchronous](<https://devfeed.tech/tags/asynchronous.md>), [await](<https://devfeed.tech/tags/await.md>), [backpressure](<https://devfeed.tech/tags/backpressure.md>), [channel](<https://devfeed.tech/tags/channel.md>), [code](<https://devfeed.tech/tags/code.md>), [compiler](<https://devfeed.tech/tags/compiler.md>), [coroutines](<https://devfeed.tech/tags/coroutines.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flowable](<https://devfeed.tech/tags/flowable.md>), [flowablesubscriber](<https://devfeed.tech/tags/flowablesubscriber.md>), [interoperation](<https://devfeed.tech/tags/interoperation.md>), [kotlin](<https://devfeed.tech/tags/kotlin.md>), [kotlin-coroutines](<https://devfeed.tech/tags/kotlin-coroutines.md>), [notify](<https://devfeed.tech/tags/notify.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>), [stream](<https://devfeed.tech/tags/stream.md>), [subscription](<https://devfeed.tech/tags/subscription.md>), [suspend](<https://devfeed.tech/tags/suspend.md>)

## AI overview

This tutorial explains how to make RxJava work with Kotlin Coroutines. It introduces a suspendable emitter and a coroutine-based Flowable generator that can suspend emission until downstream demand is available.

## Source excerpt

Introduction Writing imperative-looking code with Kotlin Coroutines is certainly an attractive property of it, but I'd think things can get quite convoluted pretty fast once, for example, Selectors are involved. I haven't gotten there to look at what Selectors are, I only read that they can help you implement a flatMap like stream combiner. We are not goind to do that now, RxJava can do it for us after all. However, the reasonable question arises: if I have a coroutine generator, a coroutine transformation or simply want to receive items from a Flowable, how can I make RxJava work with these coroutines? Easily with the combined magic of Kotlin Coroutines and RxJava coroutines! Suspendable Emitter A generator is a source-like construct that emits items followed by a terminal signal. It should be familiar from RxJava as the Flowable.generate() operator. It gives you a FlowableEmitter and the usual onNext, onError and onComplete calls on it. One limitation is that you can call onNext only once per invocation of your (Bi)Consumer lambda that receives the emitter. The reason is that we can't block a second call to onNext and we don't want to buffer it either; therefore, RxJava cooperates with the developer. Compiler supported suspension and state machine built by it, however, allow us to prevent a second call from getting through by suspending it until there is a demand from the downstream, which then resumes the coroutine where it left off. Therefore, we can lift the single onNext requirement for our Coroutine-based generator. So let's define the SuspendEmitter interface interface SuspendEmitter<in T> : CoroutineScope { suspend fun onNext(t: T) suspend fun onError(t: Throwable) suspend fun onComplete() } By extending the CoroutineScope, we provide useful infrastructure (i.e., coroutineContext, isActive) to the block that will target our SuspendEmitter. One can argue that why use onError and onComplete since a coroutine can throw and simply end. The reason is that this w