Andrey Aleev

Today I’m gonna tell how concatMap helps to transform an existing sequential dependent synchronous chain of methods into a reactive and multithreaded one.
Suppose we get a boolean value from a presenter, and perform some UI event:

  ...
  boolean state = mPresenter.getBooleanState(context)
  if(state) {
      showViewA();
    } else {
      showViewB();
  }

The class, that provides data, Presenter in our case, has a dependent method structure, the second depends on the result of the first:

public class MyPresenter<MyView> {

  ...

  private int getIntValue(Context context) {
      int retValue = someCalculationMethod(context);
      return retValue;
  }
  public boolean getBooleanState(Context context) {
      int intValue = getIntValue(context);
      return performSomeCalculationWith(intValue);
  }
}

When someCalculationMethod executes immediately, it might be run on the main thread. If it for example, server request, and we may expect some time delay, then it should be executed in a background thread.
With Rx we can easily achieve it, preliminary transformed methods getIntValue and getBooleanState into Observable:

  private Observable<Integer> getIntValue(Context context) {
      return Observable.fromCallable(() -> {
         int retValue = someCalculationMethod(context);
         return retValue;
      });
  }

  private Observable<Boolean> getBooleanState(Context context, Integer intValue) {
     return Observable.fromCallable(() -> {
        boolean retValue = performSomeCalculationWith(intValue);
        return retValue;
     });
  }

So, how do we get Observable from the getBooleanState without having getIntValue performed? All magic on merging these two Observables is made by concatMap operator:

  private Observable<Boolean> rxGetBooleanState(Context context) {
      Observable integerObservable = getIntValue(context)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());

      Observable retObservable = integerObservable
        .concatMap(intValue -> getBooleanState(context, intValue));
      return retObservable;
  }

ConcatMap works similarly to flatMap. Important distinctive feature is it keeps elements order.

Let’s check docs:

Returns a new Observable that emits items resulting from applying a function that you supply to each item emitted by the source Observable, where that function returns an Observable, and then emitting the items that result from concatenating those resulting Observables.

And look at the concatMap operator scheme:

concatMap applies a function (Observable) that you supply to each item emitted by the original Observable, and then merges the results of that function applied to every item emitted by the original Observable, thus creating a new Observable saving original order from first Observable.

And now we’re only need to call our method from appropriate context:

Disposable disposable = mPresenter.rxGetBooleanState(this)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { state ->
            if (required) {
                showViewA();
            } else {
               showViewB();
            }
        }

Update
Actually, the easiest way to execute dependant observables is to use .flatMap operator, as described here: https://github.com/ReactiveX/RxJava/issues/442

getIntValue(context).flatMap( intValue -> {
   return getBooleanState(context, intValue)
})

Today I’ll tell you about a convenient reactive approach to handle EditText input. As an example, we will take a search string in a SearchView.

The task

Search usually takes a while. No matter if we requesting data from the server, or read the local database. User may input letters in the textfield much faster than the requests would work off. Therefore we need to settle some time threshold not making any requests more frequent than necessary. Besides, we may need to update some UI changes in real-time, without those time gaps.

Solution

Without Rx in Android we can use Runnable and Handler, which is not handy. But thanks to the RxBinding we may use the debounce operator and get an elegant solution:

RxSearchView.queryTextChanges(searchView) 
 .debounce(500, TimeUnit.MILLISECONDS) // delay for 500 ms
 .subscribe(query -> mPresenter.searchRequest(query));

Now, regardless of input speed, requests to the server will be performed no more often than twice a second.
But, what should we do, if we need to update UI on every letter input? For example, hint needs to be changed. Using the code above we would have debounced UI updates with 500 ms, what is not quite user-friendly.

Fortunately, in Rx there is the possibility to broadcast to multiple subscribers. In order to use it we need to call .share() on our Observable and voila!

Observable<String> sharedTextChanges = RxSearchView.queryTextChages(searchViw).share()
 
sharedTextChanges 
 .debounce(500, TimeUnit.MILLISECONDS) // use debounce 
 .subscribe(query -> mPresenter.searchRequest(query)); 
 
sharedTextChanges 
 .subscribe(query -> mPresenter.updateUI(query));

We subscribed to Observable twice: with and without debounce. Now UI refreshes on every letter input, whereas server request being performed no more frequent than 500 ms.

What’s under the hood?

Operator .share() is an wrapper for .publish().refcount(). They allow to “share” items emitted by a stream. Let’s look deeper.

Operator.publish( ) — transforms Observable into ConnectableObservable.

“ConnectedObservable” is an Observable, which is not emitting data until someone wouldn’t call .connect() operator on it.

Operator .refcount() controls multiple subscribers. According to the docs,

Returns an Observable that stays connected to this ConnectableObservable as long as there is at least one subscription to this ConnectableObservable.

refcount() operator knows how many subscribers are subscribed to the Observable and doesn’t disconnect from the source ConnectedObservable until all Observables are unsubscribed.

That’s all. Very effective and elegant, as always with Rx. Hope this post was useful for you!