RxJava Tidbits #3: Recursion using Subjects

I recently had to implement a simple retry logic for one of our endpoints. I initially solved it using oldskool method recursion, but then replaced it with a nicer RX-solution using a BehaviorSubject.

Let’s say our endpoint returns the following JSON (highly simplified):

{
  "result": { ... },
  "nextUrl": "http://www.next.url"
}

The logic is as simple as “keep hitting the next URL until we get a result object”.

Our POJO representing that JSON response would look like this:

public class Response {
    private final Object result;
    private final String nextUrl;
    public Response() {}
    public Response(Object result, String nextUrl) {
        this.result = result;
        this.nextUrl = nextUrl;
    }

    public Object getResult() {
        return result;
    }

    public String getNextUrl() {
        return nextUrl;
    }
}

Here is a solution based on method recursion:

private static final String URL = "http://www.url.com";
public static void main(String[] args) {
    downloadFromNetwork(URL)
        .subscribe();
}
private static Observable<Response> downloadFromNetwork(String url) {
    return networkClient.get(url)
            .flatMap(new Func1<Response, Observable<Response>>() {
                @Override
                public Observable<Response> call(Response response) {
                    if (response.getResult() == null && response.getNextUrl() != null) {
                        return downloadFromNetwork(response.getNextUrl());
                    } else {
                        return Observable.just(response);
                    }
                }
            });
}

This works just fine. We keep calling downloadFromNetwork() recursively with the next URL until the result is non-null.

Now let’s see how we can achieve the same result using a BehaviorSubject:

private static final String URL = "http://www.url.com";
public static void main(String[] args) {
    final BehaviorSubject<String> subject = BehaviorSubject.create();
    subject.onNext(URL);
    subject.flatMap(new Func1<String, Observable<Response>>() {
        @Override
        public Observable<Response> call(String url) {
            return networkClient.get(url);
        }
    }).flatMap(new Func1<Response, Observable<Response>>() {
        @Override
        public Observable<Response> call(Response response) {
            if (response.getResult() == null && response.getNextUrl() != null) {
                subject.onNext(response.getNextUrl());
                return Observable.empty();
            } else {
                subject.onCompleted();
                return Observable.just(response);
            }
        }
    }).subscribe();
}

A BehaviorSubject emits the most recent item it has observed as well as all subsequent items once you subscribe to it. That allows us to emit the initial URL via subject.onNext(URL) before actually subscribing to the Subject.

Once we receive a response, instead of recursively calling ourselves again, we can now just pump the next URL into the Subject via subject.onNext(response.getNextUrl()) which will trigger the next network call repeatedly up until we get our desired non-null result.