RXJava BehaviorSubject Asynchronous Initialization

Question

In my Android app, I use the BehaviourSubject “to get data” from the data provider to my user interface and other services that need data. For example, suppose these are messages for our user.

Whenever data (such as messages) is updated, the data provider makes a “long” (so-called “slow”) network call to retrieve the message and feed it to the object, invoking the topic onNext(data) in this way, “transmitting” the update user interface and other subscribers.

This works well, but I had a problem initializing the object, or in another way setting the initial value of the object at the beginning of the application.

I know that I can set the initial value through BehaviorSubject.create(initialValue) , but since initialValue is the result of a network call, this blocks the initialization of the object.

I am currently doing the following in my data provider:

 BehaviorSubject<Data> subject = BehaviorSubject.create(); getDataFromNetwork().subscribe(data -> subject.onNext(data)); 

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

Question: The above wiring design, observed, which is updated manually from the network in BehaviourSubject, somehow feels wrong / not elegant. Is there a better way to initialize a BehaviourSubject with another observable?

I think of something like: BehaviorSubject.create(Observable obs) or in my case BehaviourSubject.create(getDataFromNetwork()) that will set up the object, leave it blank until Observable emits something and then "pushes" this is something to their subscribers.

Answer 1

What is wrong is that you are using a theme. In Rx, this is a general recommendation not to use items (as much as possible), and you will be surprised how much you can achieve without them.

In your case, you just need to set the observable instead of the object:

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

It will work the same way you try to trick BehaviorSubject , but completely Rx.

If you want to update the request, switch is the operator that you will probably need, for example. assuming refreshObs is observable, which indicates an update:

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