How to use the state of one observable to skip values of another observable?

java, rx-android, rx-java

Solution

Your pictures of the desired output is not correct. In your implementation `combineLatest` is combining a `null` into the stream :

interval       1--2--3--4--5--6--7--8--9  
checkbox       0-----1--------0-----1---
combineLatest  N--N--3--4--5--N--N--8--9
filter(NonNull)------3--4--5--------8--9 

IMHO, using `null` as a signal is not good in Rx Stream, developers can easily fall into `NullPointerException`.

To avoid the use of `null`, there are two measures. The first one is to transform the result to a Pair, and apply filter & map later.

A very simple `Pair` Class:

public class Pair<T, U> {
    T first;
    U second;

    public Pair(T first, U second) {
        this.first = first;
        this.second = second;
    }
}

Then the whole implementation would like this:

Observable.combineLatest(
        RxCompoundButton.checkedChanges(checkBox)
        , Observable.interval(1, TimeUnit.SECONDS)
        , (isChecked, intervalCounter) -> new Pair<>(isChecked,intervalCounter)
).filter(pair -> pair.first)
 .map(pair -> pair.second)  // optional, do this if you only need the second part

The data flow:

interval          1      2      3      4
                  |      |      |      |
checkbox        F |  T   |   F  |  T   F
                  |  |   |   |  |  |   |
combineLatest    F1  T1  T2  F2 F3 T3  F4
                     |   |         |
filter(first=T)      T1  T2        T3
                     |   |         | 
map(second)          1   2         3

Or if you can Java 8 in your project, use `Optional` to avoid `null`, which is very similar to your solution but it gives the awareness for others developers that the Stream signal is `Optional`.

Observable<Optional<Integer>> o1 = Observable.combineLatest(
        RxCompoundButton.checkedChanges(checkBox)
        , Observable.interval(1, TimeUnit.SECONDS)
        , (isChecked, intervalCounter) -> {
            if (!isChecked) {
                return Optional.empty();
            } else {
                return Optional.of(intervalCounter);
            }
        }
)

Observable<Integer> o2 = o1.filter(Optional::isPresent).map(Optional::get)

Some easier approach which is not same with `combineLatest` but identical to your desired result pictures.

// Approach 1
Observable.interval(1, TimeUnit.SECONDS).filter( i -> checkBox.isEnabled())

// Approach 2
Observable.interval(1, TimeUnit.SECONDS)
  .withLatestFrom(
    RxCompoundButton.checkedChanges(checkBox),
    (isChecked, intervalCounter) -> {
            if (!isChecked) {
                return Optional.empty();
            } else {
                return Optional.of(intervalCounter);
            }
     }).filter(Optional::isPresent).map(Optional::get)

Problem

Original source