Understanding streams of data and multiple subscribers (using retrofit)

3
votes

Say I have 2 Observables (A & B) that are essentially network calls (using Retrofit to give context).

The current flow of the app is as follows:

  • A & B are kicked off at about the same time (asynchronously).
  • B is executed 0 or more times on user interaction

I have 3 different scenarios that I want to listen for given these 2 observables/api calls.

  1. I want to know immediately when Observable A completes
  2. I want to know immediately when Observable B completes
  3. I want to know when both have completed

First off, is this a good use case for RxJava?

I know how to do each scenario individually (using zip for the last), though I don't know how to do all of them simultaneously.

If I subscribe to Observable A, A begins. If I subscribe to B, B begins. If A & B complete before I subscribe to zip(a, b), I could miss the event and never actually see this complete, right?

Any general guidance would be appreciated. My RxJava knowledge is pretty thin :P

1
androidretrofitrx-javarx-android

1 Answers

5
votes

You can achieve this using three different observable, one for each of your case.

As you'll have to share states between each observables, you'll have to convert retrofit cold observables to hot observable. (see here for more information on this topic)

ConnectableObservable a = service.callA().publish(); 
ConnectableObservable b = service.callB().publish();

a.subscribe((e) -> { /* onNext */ }, (ex) -> {/* onError */},  () -> {/* when A is completed */ });
b.subscribe((e) -> { /* onNext */ }, (ex) -> {/* onError */},  () -> {/* when B is completed */ });
a.mergeWith(b).subscribe((e) -> { /* onNext */ }, (ex) -> {/* onError */},  () -> {/* when A and B are completed */ });

a.connect(); // start subscription to a
b.connect(); // start subscription to b

Do not share an object between onCompleted methods or you'll have to deal with concurrencies issues.