First look at RxJava

I try RxJava for the first time and immediately close the tab. Too many new words. Observable, Observer, Subscriber, Subject, Scheduler, Operator. What is this.

Two weeks later a colleague shows me a real example and something clicks.

The problem it solves

We have an API call that needs to run on a background thread and update the UI when done. The standard way is AsyncTask.

new AsyncTask<Void, Void, User>() {
    @Override
    protected User doInBackground(Void... params) {
        return api.getUser(id);
    }

    @Override
    protected void onPostExecute(User user) {
        nameTextView.setText(user.getName());
    }
}.execute();

This works for one request. But now chain two requests. Then add error handling. Then add a loading state. The AsyncTask code becomes a mess very fast.

Observables

In RxJava everything is a stream. An Observable emits items. A Subscriber listens.

Observable.just("hello", "world")
    .subscribe(item -> Log.d("TAG", item));

That's the simplest case. The real power is the operators.

Operators

map transforms each item:

Observable.just(1, 2, 3)
    .map(n -> n * 2)
    .subscribe(n -> Log.d("TAG", String.valueOf(n)));
// 2, 4, 6

flatMap is the one that takes time to understand. It transforms each item into another Observable, then flattens everything:

api.getUserIds()
    .flatMap(id -> api.getUser(id))
    .subscribe(user -> showUser(user));

Get a list of ids, then for each id fetch the user. Concurrent by default. In AsyncTask this is many nested calls.

filter removes items you don't want. zip combines two streams. debounce waits for the user to stop typing before triggering a search. These are the ones I use every week.

Schedulers

This is where it gets useful for Android:

api.getUser(id)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        user -> showUser(user),
        error -> showError(error)
    );

subscribeOn says where the work runs. observeOn says where the result is delivered. Two lines replace all the thread switching code.

The hard part

Error handling with RxJava is different. Errors travel through the stream. If you don't handle them in the subscriber the app crashes.

Also memory leaks. If the Activity is destroyed and the Observable is still running, your subscriber holds a reference to the Activity. You have to unsubscribe manually or use a library like RxLifecycle.

@Override
protected void onDestroy() {
    super.onDestroy();
    compositeSubscription.unsubscribe();
}

Easy to forget. Causes crashes. You learn the hard way.

Is it worth it

For simple apps, maybe not. AsyncTask and callbacks are easier to understand.

For anything with complex async flows, yes. Searching as you type, combining multiple API calls, polling, all of these become clean with RxJava.

The learning curve is steep. But after a few weeks the operators start feeling natural and you stop thinking in callbacks.