Don't re-execute Retrofit call if it's still in progress using RxJava 2

Question

My goal is a method that will execute a network call unless the network call is already in progress in which case the caller subscribes to the results of the existing call.

Here is what I have, it mostly works:

private AsyncSubject<Stuff> asyncSubject;

public Observable<Stuff> getStuff() {
    if (asyncSubject == null) {
        asyncSubject = AsyncSubject.create();
        asyncSubject
            .doAfterTerminate(new Action() {
                @Override
                public void run() throws Exception {
                    asyncSubject = null;
                }
            })
            .subscribe();
        retrofitApi.getStuff()
            .subscribe(asyncSubject);
    }
    return asyncSubject
        .someOperatorsHere();
}

This mostly works because asyncSubject nulls itself out on terminate so that any subsequent calls will re-execute the network request.

But there is a bug, there's a window between if (asyncSubject == null) and return asyncSubject where doAfterTerminate can execute and I get a null pointer.

Any ideas how to fix the bug. Maybe there is a more elegant Rx way to do this? Or synchronized block. Or a way to query Retrofit for network progress.

Answer 1

This solution shares the subscription between ongoing subscribers.

private Observable<Stuff> stuff = retrofitApi.getStuff().share();

public Observable<Stuff> getStuff() {
    return stuff
        .someOperatorsHere();
}

Answer 2

There are couple of possible solutions, I'll just describe the one that is probably simplest in your case.

You have 2 problems: race condition, and nullifying and recreating subject, which is rather unsafe.

What you can do is save your Subscription here

retrofitApi.getStuff().subscribe(asyncSubject);

And instead of checking if (asyncSubject == null), you can check if (s.isUnsubscribed()).

So your code will look more or less like this:

private AsyncSubject<Stuff> asyncSubject = AsyncSubject.create();
private Subscription subscription = Subscriptions.unsubscribed();

public Observable<Stuff> getStuff() {
    if (subscription.isUnsubscribed()) {
        subscription = retrofitApi.getStuff()
            .subscribe(asyncSubject);
    }
    return asyncSubject
        .someOperatorsHere();
}