Different naming conventions. Subscription was renamed to Disposable. Instead of .unsubscribe() you now call .dispose() .

Also CompositeSubscription became CompositeDisposable , the same .clear() method is used to dispose all the contained disposable.

The more the merrier

Because our app was using RxJava1 in a lot of places we decided to do the migration gradually.

Rx1 and Rx2 have two different package names rx.* and io.reactivex.* this means you can have both versions in one project. This is exactly what we did.

This gave us some flexibility in the sense that the migration didn’t have to happen all at once. We also were using external libraries that were handing us Rx1 Observables so we had to keep that dependency for those cases as well (until we find alternatives to remove those dependencies).

We did the same for RxAndroid and RxBindings and our retrofit call adapter as well:

"io.reactivex.rxjava2:rxjava:${rxJava2Version}"

"io.reactivex.rxjava2:rxandroid:${rxAndroid2Version}"

"com.jakewharton.rxbinding2:rxbinding:${rxBindings2Version}" "com.squareup.retrofit2:adapter-rxjava2:${retrofitVersion}"

Learnings and tips

The main learning I get from this is the use of the primitive types. They allow you to have a cleaner code and better vision of what your code is doing. Also it ensures correctness of what you’re writing.

For example if you try to apply a .filter() to a Single it will give you a Maybe back and forces you to think of the case where there is no item emitted. This makes sense because if you filter a single value it will either give you the result (pass the filter) or simply completes without the filtered item.

A typical Maybe use case for us was when we wanted to show some cached data to the user but if we didn’t have anything cached, make a network request to get the data. This is how it looked like before:

override fun getContent(): Observable<SomeContent?> {

return Observable.fromCallable(this::loadContentFromPrefs)

.flatMap { cachedContent: SomeContent? ->

when {

cachedContent == null || cachedContent.isEmpty() -> loadContentFromApi()

else -> Observable.just(cachedContent)

}

}

} private fun loadContentFromPrefs(): SomeContent? {

content = gson.fromJson(prefs.getString(KEY_CONTENT, null), SomeContent::class.java) return content

}

Here we’re getting the content from the shared preferences, if it’s null or empty we make a network request otherwise we just emit it. So now our function returns an Observable<SomeContent?> which means we don’t know how many items will be emitted and they could be null (we know they won’t but since loadContentFromPrefs() could emit null into the stream we had to keep the nullable type)

This is the version with RxJava2:

override fun getContent(): Single<SomeContent> {

return loadContentFromPrefs()

.switchIfEmpty(SingleSource {

loadContentFromApi()

});

} private fun loadContentFromPrefs(): Maybe<SomeContent> {



return Maybe.create({ sub ->

val prefContent = gson.fromJson(prefs.getString(KEY_CONTENT, null), SomeContent::class.java)



if (prefContent != null) {

sub.onSuccess(prefContent)

} else {

sub.onComplete()

}

})

}

This version does exactly the same thing, except we get a Single<SomeContent> as return type so we know this stream will only emit one item (and it won’t be null), or emit an error. The small trick here is the use of the switchIfEmpty() operator, which will “flatmap” the Maybe into a Single in case the Maybe completed without emitting an item. Basically if our Maybe emits an item it will go through, otherwise it will call our loadContentFromApi() Single.

Also working with Maybe there is the flatMapSingleElement() which will act as a “flatmap” on the single element emitted by the Maybe if any.

Safe use of the .create() methods. Prior to Rx2 the .create() methods of RxJava was something we would avoid to use because it wasn’t the right way of creating Observables, we would use .just() , .from() , .fromCallable() most of the time. This is a thing of the past and we can now easily create custom streams with .create() .

Unit testing

To be able to do unit testing with RxJava we took the habit of changing threads on where things were executed. Rx2 still offers this possibility, the only difference is that the Schedulers.immediate() no longer exist and is replaced by Schedulers.trampoline() . So when unit testing we are replacing both Schedulers.io() and AndroidSchedulers.mainThread() with Schedulers.trampoline() .

@Before

fun setup() {

// RxJava2 override

RxJavaPlugins.reset();

RxJavaPlugins.setIoSchedulerHandler({ scheduler ->

Schedulers.trampoline()

}); // RxAndroid override

RxAndroidPlugins.reset();

RxAndroidPlugins.setMainThreadSchedulerHandler({ scheduler ->

Schedulers.trampoline()

});

}

Conclusion

The transition wasn’t as bad as I thought it would be. Most of it was replacing Observable with Single , then making sure we’re using the version 2 of RxBindings and RxAndroid and replacing Subscription with Disposable .

There was some obstacle on the way with nulls in the streams that forced us to make use of the new primitive types, which is a good thing. Now when we look at our function signatures we know how the returned stream is supposed to behave and what type of result is expected (one item, only completion..).

Some more readings:

- https://github.com/ReactiveX/RxJava/wiki/What%27s-different-in-2.0

- https://blog.kaush.co/2017/06/21/rxjava1-rxjava2-migration-understanding-changes/

- https://artemzin.com/blog/reply-to-kaushik-gopals-aricle-rxjava-1-rxjava-2-understanding-the-changes/