Introduction
In this article, we will compare how to convert streams of streams using the Operators.
Common Operators
Several RxJava operators are available in the standard Kotlin library as part of the Kotlin Collections . Here you can see a table comparing simple operators.
These operators convert the stream in the same way, albeit with a few different names: for example, skip in RxJava is called drop in Coroutines.
Create Your Own Operator
Some operators are not part of the Kotlin Collections. However, you can easily create them yourself.
We can make the range RxJava operator perform with a few lines of code and a simple for loop.
1 2 3 4 5 6 7 8 | fun range( context: CoroutineContext, start: Int, count: Int ) = publish(context) { for (x in start until start + count) send(x) } |
Some other operators require more effort. In the example below, you can see the implementation of the RxJava Completable.zip operator, which takes two blocks of code and waits for both blocks to finish.
1 2 3 4 5 6 7 8 9 10 11 | suspend fun zip( context: CoroutineContext, block: () -> Unit, block2: () -> Unit ) { val deferred1 = async(context) { block() } val deferred2 = async(context) { block2() } deferred1.await() deferred2.await() } |
If you notice, we pass a CoroutineContext as an argument. We do that so we can easily cancel the operator by calling .cancel () on the Job of that context as mentioned in Part 2.
Complex Operators
What about the even more complex RxJava operators like debounce?
You can find the debounce as an extension function on ReceiveChannel. RxJava’s timeout is equivalent in Kotlin Coroutines to withTimeoutOrNull, etc.
Similarities and Differences
We see that most of the operators are available in both libraries, and if not, you can easily build them.
The only difference that can be seen in these two libraries is when you apply those operators.
Whereas with RxJava, you can apply the operators before subscribing to the stream, you have to do that after opening the registration in Coroutines. Let’s see how to map an element in RxJava:
1 2 3 4 5 6 | rxObservable .map { result -> map(result) } .subscribe({ consumeResult(it) }) |
And now how do we do that in Coroutines:
1 2 3 4 | broadcastChannel .openSubscription() .map { result -> map(result) } |
In Coroutines we have to do it after opening the registry because the map in Coroutines is an extension function on ReceiveChannel <E>. This is the case for other operators like filter, drop, etc.When calling openSubscription (), a SubscriptionReceiveChannel <T> object that extends from ReceiveChannel <E> is returned.
As such, Coroutines requires extra work if you want multiple Observers to apply the same operators. Of course you can! But it requires more code.