Our Top Nine Learnings from Migrating from RxJava 1 to RxJava 2
by Roman Zavarnitsyn,
Software Engineer Android @Runtastic
At Runtastic we’re using RxJava to establish functional and reactive programming patterns. Modern mobile apps and RxJava are a great fit. We had been using RxJava 1 for a while, but recently we’ve switched major parts of our codebase to RxJava 2.
RxJava2 introduces breaking changes. The official documentation outlines all the changes between RxJava 1 and 2. In this blog post, we want to outline the biggest learnings we had when migrating to RxJava 2. We’ll use the Groups feature of the Runtastic Running & Fitness app as an example throughout this blog post.
1. You can use RxJava and RxJava 2 side by side
If you look at the RxJava 2 dependency path, you can see that it has a new artifact and new package name:
io.reactivex.rxjava2:rxjava2:$version_number
An important thing here is that RxJava 2 is not built on top of the previous version. Due to the new Reactive Streams specification, the authors decided to rewrite RxJava 2 completely. This is a great benefit for us developers because we can use both versions of RxJava in one project. This makes our migration much smoother and easier. Migrating everything at once would be almost impossible for big projects, but because of the difference in package names, we could change our code gradually.
2. Use RxJava2Interop for a smooth transition
RxJava2Interop is a very handy library from one of the main RxJava contributors, which allows you to convert between 1.x and 2.x reactive base types. At Runtastic we use the Model-View-Presenter architecture pattern. Therefore, if we migrate our Model Observables, we’d need to do the same thing for our Presenter, otherwise it wouldn’t compile at all. But thanks to interoperability, we are able to change our code class by class, or even method by method. As an example, let’s look at our Groups functionality migration:
Model
public Single < Group > createGroup(final String name, final String description) {
return Single.fromCallable(() - >{
if (!NetworkInteractorFactory.getNetworkInteractor(context).isConnected()) {
throw new NoConnectionException();
}
return GroupsHelper.createGroup(context, name, description);
});
}Presenter
compositeSubscription.add(RxJavaInterop.toV1Single(groupsRepo.createGroup(name, description))
.subscribeOn(Schedulers.io())
.observeOn(viewScheduler)
.subscribe(group -> {
view.openGroupDetails(group);
}));Here, in Model, we used Single from RxJava 2 and RxJava2Interop to convert it to v1 Single. It helped us a lot because we were able to compile and run our tests after every method migration. It’s much easier to handle any breaking changes in only one method than in the whole module/project.
3. Don’t use Flowable everywhere
Backpressure has always been a problem, especially if you have some hot sources or you’re using a custom create operator. There are plenty of different articles and questions about what to use and in which cases, but there is no single source of truth. After spending some time investigating this, we ended up deciding to use Flowable only when we really need backpressure handling enabled. This is not the case for everyday Android development things like network requests or database querying.
You could face MissingBackpressureException when you’re processing user input events (touch events while drawing, audio stream while speaking to a microphone), but we don’t have such features in our apps. So we’re good to go with Observable, as it’s more lightweight.
4. Don’t hesitate to use Maybe, Single and Completable
Firstly, let’s talk about Maybe, which is a new reactive type in RxJava. It either succeeds with an item (onSuccess), completes with no items (onComplete), or errors (onError). You should consider it as Optional.
To be honest, we hadn’t used any other reactive types than Observable in our code. But during the migration, we decided to inspect our reactive types usage as well, because they can give more context and each of them has a different set of operators we could use in appropriate situations to make our code more readable.
At Runtastic we’re using Retrofit for network communication. So assume we have a GroupsEndpoint which can:
- Return a group from a remote storage
- Save a group to this storage
- (Possibly) return a list of joined groups
For the first case, we could use Single for requesting a group because the source is supposed to emit only one item and get terminated.
For the second case, it’s reasonable to use Completable, as it’s not supposed to return anything, it just saves the group and successfully completes, followed by onComplete callback.
And finally, for the third case, we would use Maybe in order to be more explicit about the expected return value. If the user hasn’t joined any groups, just the onComplete callback would get fired.
Given that explanation, our GroupsEndpoint looks like:
public interface GroupsEndpoint {
@GET(...)
Single<Group> getGroupById(@Path("groupId") String groupId);
@PUT(...)
Completable saveGroup(@Body Group group);
@GET(...)
Maybe<List<Groups>> getJoinedGroups();
}Now it feels better and gives more context and details to other developers, so they have fewer ways to interpret and handle our interface incorrectly.
5. Deal with Null emissions
In RxJava 2, it’s not allowed anymore to emit null values. So the following code will produce NullPointerException immediately:
Observable.just(null); subject.onNext(null);
We found 3 possible ways of dealing with nulls:
- Avoid null emission. You could throw a custom descriptive Exception instead, which will fire onError callback with that Exception as a parameter. We did it at several places in our code, and it was a really good thing to rethink the business logic and get rid of some unnecessary nulls.
- If you’re interested in knowing if the returned value was null, you can change your source to emit Optional from Java 8 or Guava library. And then you just have to check for value presence by using the isPresent method.
- As mentioned in the section above, you can use Maybe if you’re dealing with a network request which could return an empty body. But there is also an option to implement your own Maybe source using the create operator of it:
Maybe<List<Groups>> maybeJoinedGroups = Maybe.create(emitter -> {
try {
List<Group> groups = getJoinedGroups();
if (groups != null && !groups.isEmpty()) {
emitter.onSuccess(groups);
} else {
emitter.onComplete();
}
} catch (Exception e) {
emitter.onError(e);
}
})You can choose whichever way you want, but you should definitely keep in mind and possibly rethink some of your business logic.
6. Use the subscribeWith method when you have a reusable Observer implementation
During migration, we found some duplicated code in our onNext/onError callbacks. This led us to implement a custom Observer to reuse it and avoid code duplication:
public class GroupsObserver extends DisposableObserver {
…
}But then we realized unexpectedly that the subscribe (Observer observer) method returned void and we weren’t able to dispose the Observer.
The solution was simple: we just used the new dedicated subscribeWith method that returns the Observer instance which has been passed as a parameter:
compositeDisposable.add(groupsRepo.getJoinedGroups() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(groupsObserver));
7. Defer execution of function with the fromCallable method
We know that fromCallable has been introduced earlier than RxJava 2, but we considered it necessary to point it out. While inspecting our code for RxJava 1 usages, we were faced with code like this:
Single.defer(() - > {
if (!NetworkInteractorFactory.getNetworkInteractor(context).isConnected()) {
return Single.error(new NoConnectionException());
}
try {
return Single.just(GroupsHelper.createGroup(context, name, description));
} catch (Exception e) {
return Single.error(e);
}
});That was a very old approach to deferring the execution of function until the observer subscribed. The new method fromCallable was introduced in the RxJava 1.0.15 version, but we didn’t change it back then. By now switching to the new method, our code has been changed to:
return Single.fromCallable(() - > {
if (!NetworkInteractorFactory.getNetworkInteractor(context).isConnected()) {
throw new NoConnectionException();
}
return GroupsHelper.createGroup(context, name, description);
});It actually deals with checked exceptions and invokes the onError method, so there is no need to try-catch anymore. Looks much clearer and nicer now.
8. Use the new test method for tests
There is a new convenient method that has been added in RxJava 2 which is called test. What it does is return either TestSubscriber or TestObserver. Thanks to this method, you can now test your sources directly, without introducing TestSubscriber/TestObserver instances. Given that most TestSubscriber/TestObserver methods return the instance itself, we can chain assert calls like this:
groupsRepo.getJoinedGroups() .map(group -> addUsersToGroup(group)) .test() .assertNoErrors() .assertComplete() .assertResult(dummyGroupsWithUsers);
Now it looks very readable and consistent. In addition, it’s very convenient and easy to write such chains.
9. Use the trampoline scheduler for instant results emission in tests
Many of you might have used the popular RxSchedulersOverrideRule for immediate execution of work in tests with RxJava as we did. Now it won’t work since immediate does not exist in RxJava 2 anymore. Instead, we should use trampoline, which does almost the same thing as immediate (invokes task on a current thread in a blocking way), but it also waits till all previously tasks are completed.
So we changed our RxSchedulersOverrideRule class implementation to this one:
public class RxSchedulersOverrideRule implements TestRule {
private final Function <Callable<Scheduler>, Scheduler> mRxAndroidSchedulersHook = schedulerCallable - > Schedulers.trampoline();
private final Function <Scheduler, Scheduler> mRxJavaSchedulersHook = scheduler - > Schedulers.trampoline();
@Override
public Statement apply(final Statement base, Description description) {
return new Statement() {
@Override
public void evaluate() throws Throwable {
RxAndroidPlugins.reset();
RxAndroidPlugins.setInitMainThreadSchedulerHandler(mRxAndroidSchedulersHook);
RxJavaPlugins.reset();
RxJavaPlugins.setIoSchedulerHandler(mRxJavaSchedulersHook);
RxJavaPlugins.setNewThreadSchedulerHandler(mRxJavaSchedulersHook);
RxJavaPlugins.setComputationSchedulerHandler(mRxJavaSchedulersHook);
base.evaluate();
RxAndroidPlugins.reset();
RxJavaPlugins.reset();
}
};
}
}Conclusion
It’s not a complicated task to switch from RxJava to RxJava 2, but it does take some effort, as you need to keep things working after migration. Just keep in mind some of the points mentioned in this article, and you’re good to go. Feel free to share your takeaways and learnings in the comments section below. Happy coding!
***
