RxJava: Find out if BehaviorSubject was a repeated value or not

Question

I'm making an Android interface that shows some data fetched from the network. I want to have it show the latest available data, and to never be empty (unless no data has been fetched at all yet) so I'm using a BehaviorSubject to give subscribers (my UI) the latest available info, while refreshing it in the background to update it.

This works, but due to another requirement in my UI, I now have to know whether or not the published result was gotten fresh from the network or not. (In other words, I need to know if the published result was BehaviorSubject's saved item or not.)

How can I achieve this? If I need to split it up into multiple Observables, that's fine, as long as I'm able to get the caching behavior of BehaviorSubject (getting the last available result) while also being able to tell if the result returned was from the cache or not. A hacky way I can think of to do it would be to check if the timestamp of the response was relatively soon, but that'd be really sloppy and I'd rather figure out a way to do it with RxJava.

Answer 1

As you mentioned in the question, this can be accomplished with multiple Observables. In essence, you have two Observables: "the fresh response can be observed", and "the cached response can be observed". If something can be "observed", you can express it as an Observable. Let's name the first one original and the second replayed.

See this JSBin (JavaScript but the concepts can be directly translated to Java. There isn't a JavaBin as far as I know, for these purposes).

var original = Rx.Observable.interval(1000)
  .map(function (x) { return {value: x, from: 'original'}; })
  .take(4)
  .publish().refCount();

var replayed = original
  .map(function (x) { return {value: x.value, from: 'replayed'}; })
  .replay(null, 1).refCount();

var merged = Rx.Observable.merge(original, replayed)
  .replay(null, 1).refCount()
  .distinctUntilChanged(function (obj) { return obj.value; });

console.log('subscribe 1st');
merged.subscribe(function (x) { 
  console.log('subscriber1: value ' + x.value + ', from: ' + x.from);
});

setTimeout(function () {
  console.log('    subscribe 2nd');
  merged.subscribe(function (x) {
    console.log('    subscriber2: value ' + x.value + ', from: ' + x.from);
  });
}, 2500);

The overall idea here is: annotate the event with a field from indicating its origin. If it's original, it's a fresh response. If it's replayed, it's a cached response. Observable original will only emit from: 'original' and Observable replayed will only emit from: 'replayed'. In Java we would require a bit more boilerplate because you need to make a class to represent these annotated events. Otherwise the same operators in RxJS can be found in RxJava.

The original Observable is publish().refCount() because we want only one instance of this stream, to be shared with all observers. In fact in RxJS and Rx.NET, share() is an alias for publish().refCount().

The replayed Observable is replay(1).refCount() because it is also shared just like the original one is, but replay(1) gives us the caching behavior.

merged Observable contains both original and replayed, and this is what you should expose to all subscribers. Since replayed will immediately emit whenever original does, we use distinctUntilChanged on the event's value to ignore immediate consecutives. The reason we replay(1).refCount() also the merged is because we want the merge of original and replay also to be one single shared instance of a stream shared among all observers. We would have used publish().refCount() for this purpose, but we cannot lose the replay effect that replayed contains, hence it's replay(1).refCount(), not publish().refCount().

Answer 2

Doesn't Distinct cover your case? BehaviorSubject only repeats the latest element after subscription.

Answer 3

I believe what you want is something like this:

private final BehaviorSubject<T> fetched = BehaviorSubject.create();
private final Observable<FirstTime<T>> _fetched = fetched.lift(new Observable.Operator<FirstTime<T>, T>() {
    private AtomicReference<T> last = new AtomicReference<>();
    @Override
    public Subscriber<? super T> call(Subscriber<? super FirstTime<T>> child) {
        return new Subscriber<T>(child) {
            @Override
            public void onCompleted() {
                child.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                child.onError(e);
            }

            @Override
            public void onNext(T t) {
                if (!Objects.equals(t, last.getAndSet(t))) {
                    child.onNext(FirstTime.yes(t));
                } else {
                    child.onNext(FirstTime.no(t));
                }
            }
        };
    }
});

public Observable<FirstTime<T>> getObservable() {
    return _fetched;
}

public static class FirstTime<T> {
    final boolean isItTheFirstTime;
    final T value;

    public FirstTime(boolean isItTheFirstTime, T value) {
        this.isItTheFirstTime = isItTheFirstTime;
        this.value = value;
    }

    public boolean isItTheFirstTime() {
        return isItTheFirstTime;
    }

    public T getValue() {
        return value;
    }

    public static <T> FirstTime<T> yes(T value) {
        return new FirstTime<>(true, value);
    }

    public static <T> FirstTime<T> no(T value) {
        return new FirstTime<>(false, value);
    }
}

The wrapper class FirstTime has a boolean which can be used to see if any subscriber to the Observable has seen it before.

Hope that helps.