RxJava includes a wealth of operators. This article will discuss the use of common operators in Rx based on a practical scenario.

One of the things we used most in the project was Retrofit in conjunction with RxJava’s network request processing. However, there may be some special requirements or poorly designed interfaces in the project that require the client to have high precision control over the invocation of the interface, when RxJava operators come in handy.

Scenario 1: Automatic login after successful registration using flatMap

Assume that the server has two interfaces: registration and login, but the functions of the two interfaces are separated. Successful registration does not automatically log in, and the client needs to invoke the login interface again.

Now let’s analyze the scenario where you need to invoke a login after a successful login, which might be solved without using RxJava: Write another login in the login callback, which I’ll simulate using pseudo-code:

Api.register(username, password, new CallBack() { public void onSuccess(){ Api.login(username, password, New CallBack() {public void onSuccess() {public void onSuccess()}}); }});Copy the code

There is no exception handling involved here, which can be done, but it is easy to get into indenting hell, assuming that you now need to get an IP address from the network before logging in, which is another layer of callback. This code is ugly and unmaintainable, so let’s see how it works with Rx:

Api.register(account, password) .flatMap(new Function<BizResponse<UserInfo>, ObservableSource<UserLoginResponse>>() { @Override public ObservableSource<UserLoginResponse> apply(@NonNull BizResponse<UserInfo> Response) throws Exception {// Execute the login method after the registration is successful. Return api.login (Account, password); } }) .subscribe(new Consumer<UserLoginResponse>() { @Override public void accept(@NonNull UserLoginResponse UserLoginResponse) throws Exception {// Successful}}, New Consumer<Throwable>() {@override public void accept(@nonnull Throwable) throws Exception {// fail}}); new Consumer<Throwable>() {@override public void accept(@nonnull Throwable) throws Exception {// fail}});Copy the code

Here we use the flatMap operator, which is a lot like a map except that a map converts from one object to another, whereas a flatMap converts from an object to an Observable and then operates on it.

If you need to add new operations, such as asynchronously obtaining IP addresses, add this operation before the flatMap and leave the code behind the flatMap untouched. And there is a uniform exception entry — any exceptions inside are caught in the second Consumer.

Scenario 2: Using Compose to simplify the code

These two sentences are commonly added to web requests

observable.subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread());
Copy the code

Network operations are performed on the IO thread and UI operations are performed on the main thread. But every Observable writes these two lines. Can we simplify? Of course you can. This is done using the compose operator. We can define one such method in RxUtils:

/** * I/o thread executes, *.compose(RxUtils.<T> ObservableTransformer ()) */ public static <T> ObservableTransformer<T, T> applySchedulers() { return new ObservableTransformer<T, T>() { @Override public ObservableSource<T> apply(@NonNull Observable<T> observable) { return observable.subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); }}; }Copy the code

Compose (RxUtils.

applySchedulers()) is composed before the subscribe method.

Of course, this method can add more logic, such as to determine whether the data returned by an interface is abnormal, usually the server will return a status to determine:

Public static <T extends BizResponse> ObservableTransformer<T, T> applyBizSchedulers() { return new ObservableTransformer<T, T>() { @Override public ObservableSource<T> apply(@NonNull Observable<T> observable) { return observable .map(new Function<T, T>() { @Override public T apply(@NonNull final T t) throws Exception { if (t.getRcode() ! = 0) { throw new RuntimeException(t.getRmsg()); } return t; } }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); }}; }Copy the code

In the above example, if the code returned is not 0, an exception is thrown and handled in Consumer .

Scenario 3: Use ZIP to process concurrent requests

Now there is a requirement: suppose that on the user page, not only the user’s general information is displayed, but also the user’s money information is displayed. The two information corresponds to the two interfaces, and the two interfaces are accessed at the same time. If the two information is obtained successfully, the two interfaces are successfully accessed. If either information fails to be obtained, the two interfaces fail to be accessed, and you need to obtain the information again.

Without RX, this requirement is very complex to implement and can only be obtained sequentially, which increases the load time.

We can do this with ZIP in RxJava:

Observable.zip(Api.getNewsList(), Api.getBanner(), new BiFunction<List<News>, BizResponse<List<Banner>>, Pair<List<News>, List<Banner>>>() {
            @Override
            public Pair<List<News>, List<Banner>> apply(List<News> news, List<Banner> banner) throws Exception {
                return Pair.create(news, banner);
            }
        })
                .compose(RxUtils.applySchedulers())
                .subscribe(...)
Copy the code

Scenario 4: Debounce Controls the frequency

Suppose you have a search box that automatically searches for what you typed. At first glance, the requirement looks simple, just listening for the EditText input event and then calling the search interface. But if the user enters too quickly, the interface will be called too often. Is there a good way to control how often an interface is called? Debounce will do it.

RxTextView.textChanges(search)
        .debounce(1, TimeUnit.SECONDS)
Copy the code

With this code, the textChange event in EditText is called when the input interval is greater than one second, so that if the user is typing frequently the interface is called, the problem is solved.