Asnyc Initialisation of RXJava BehaviorSubject

Question

In my Android App I use a BehaviourSubject "to get data" from a data provider to my UI and other services that need the data. For the sake of an example, let's assume these are messages for our user.

Whenever a refresh of the data (e.g. messages) is triggered, the data provider will do a "long running" (aka "slow") network call to retrieve message and feed them into the subject by invoking the subject's onNext(data) operation, thus "broadcasting" the update to the UI and the other subscribers.

This works well, but I'm having a problem with the initialisation of the subject or put another way with setting the initial value of the subject on app start.

I know that I can set an initial value via BehaviorSubject.create(initialValue), but as the initialValue is the result of the network call, this would block the initialisation of the subject.

I currently do the following in my data provider's init:

BehaviorSubject<Data> subject = BehaviorSubject.create();

getDataFromNetwork().subscribe(data -> subject.onNext(data));

where getDataFromNetwork() returns an Observable for the result of the network call.

Question: The above construct of wiring the observable that refreshes from network by hand into the BehaviourSubject somehow feels wrong/not elegant. Is there a better way to initialise a BehaviourSubject with another Observable?

I'm thinking of something like: BehaviorSubject.create(Observable obs) or in my case BehaviourSubject.create(getDataFromNetwork()) that would setup the subject, leave it empty until the Observable emits something, and then "pushes" this something to its subscribers.

Answer 1

What feels wrong is that you're using a subject. In Rx it's a general recommendation not to use subjects (as much as possible), and you'd be surprised how much you can achieve w/o them.

In your case you should simply expose an observable instead of the subject:

Observable<Data> cachedData = getDataFromNetwork().publish().replay(1);

It'll work the same way you're trying to trick BehaviorSubject into, but fully Rx.

If you want to refresh the query, then switch is the operator you probably need, e.g. assuming refreshObs is an observable which ticks upon refresh:

Observable<Data> cachedData = refreshObs.map(t -> getDataFromNetwork())
                                        .switchOnNext()
                                        .publish().replay(1);