Four scenarios that are best suited for RxJava

RxJava is a very popular functional responsive programming library. It is very popular in Android development. It may be a bit difficult to get started at first, but once you understand it, you can never go back. If you don’t use RxJava to write asynchronous requests, you will feel uncomfortable.

[[201947]]

This article does not intend to talk about the basics of RxJava. If you are not familiar with RxJava, here is a good tutorial for reference: "Detailed Explanation of RxJava for Android Developers".

Next, we will introduce the four scenarios where RxJava is most suitable for use. The code examples are based on RxJava1

Scenario 1: Single request asynchronous processing

Since some time-consuming operations cannot be performed in the Android UI thread, such as network requests, large file saving, etc., asynchronous processing is often encountered in development. Our most typical usage scenario is RxJava+Retrofit to process network requests.

MyService myService = retrofit.create(MyService.class);
myService.getSomething()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::updateUI, this::showError);

To make the code look simpler, lambda expressions are used here. updateUI and showError need to be implemented in the current class, for example:

public void updateUI(Data data){
      //TODO something
}

public void showError(throwable t){
      //show error msg
}

Scenario 2: Multiple asynchronous requests are called continuously

This scenario is actually very common. When we edit user avatars, we usually need to call three requests consecutively:

1. Request the URL for uploading the avatar
2. Upload an avatar
3. Update User Information

In normal code, we need to nest callbacks step by step. The code is too long and ugly, and it is difficult to maintain. Using RxJava chain call processing code logic will be very clear

Observable.just(1)
  .map(this::task1)
  .map(this::task2)
  .map(this::task3)
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(this::updateUI, this::showError);

The fixed value 1 sent by just here has no practical meaning, but I think it is more informative.

You can also create observables using Observable.create.

Scenario 3: Merging multiple asynchronous requests

Sometimes in a project, we may encounter a situation where we need to combine the results of multiple requests and then update the UI. For example, in our project, we need to obtain notification data from multiple request addresses and then display them in chronological order on the APP. In this case, we can use RxJava's zip function to handle this.

MyService myService = retrofit.create(MyService.class);
Observable o1 = myService.getNotification1();
Observable o2 = myService.getNotification2();
Observable.zip(o1,o2, this::combiNotification)
             .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::updateUI, this::showError);


public List<Notification> combiNotification(List<Notification> n1, List<Notification> n2){
      //TODO merge notification list}

The zip function will wait for both requests to complete, call our merge method combiNotification, and then call back the method in subscribe after the merge is processed.

Scenario 4: Scheduled Polling

RxJava is also particularly suitable for processing timed polling tasks. A typical example is that an APP submits a task to the background for asynchronous processing. Assuming that the background processing takes about 1-2 minutes, we need to query the progress of the background regularly and update it to the UI. The traditional way is to use the postDelay method of Handler. If it is implemented with RxJava, it will be very concise.

Subscription subscription = Observable.interval(2, TimeUnit.SECONDS)
                .map(this::getProgress)
                .takeUntil(progress -> progress != 100)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onCompleted() {
                        //TODO finished
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(int progress) {
                         //TODO update progress
                    }
                });

We query once every 2 seconds until the progress reaches 100, and then the polling is automatically terminated.

The above RxJava methods are all asynchronous and time-consuming calls. Considering that the request has not been completed when the Activity exits, we need to cancel the RxJava call in the OnDestroy method of the Activity.

subscription.unsubscribe();