CODEWITHSUNDEEP
rx-javarx-android
dudeck
dudeck

Reputation: 413

RxJava combine multiple observables do sth after all done

I've got a method which should reload DB using multiple observables. After all it should call method setupPlace(selecetedPlace), but only once. Each observable returns different object (void, object, listOfObjects).

private void reloadDatabaseFromRest(final Place selectedPlace, final Event selectedEvent) throws ParseException {
    Observable.concat(DatabaseManager.getInstance().clearDatabase(),
            mPlaceUseCase.getPlacesListFromRestObservable()
                    .filter(places -> places != null && !places.isEmpty())
                    .doOnNext(places -> mPlaceUseCase.savePlacesToRepository(places)),
            mPlaceUseCase.saveSelectedPlace(selectedPlace),
            mEventUseCase.getEvenListObservableFromRest(selectedPlace.getId())
                    .doOnNext(eventList -> mEventUseCase.saveEventsToRepository(eventList)),
            mEventUseCase.saveSelectedEventObservable(selectedEvent))
            .takeLast(1)
            .subscribe(o -> mSplashScreenUI.setupPlace(selectedPlace));
}

But this method is doing subsribe() only on last observable and others are not invoked (am I right?). I tried using filters, zip and making stupid if(o instanceOf ArrayList) clause, but it is wrong way. Can someone help me, please?

Upvotes: 1

Views: 957

Answers (2)

dudeck
dudeck

Reputation: 413

Thank you very much @koperko! It works like a charm and this is exactly what I needed. During rewriting method I noticed that first Observable emits 6 items (clearDatabase()), so I wrapped it into Completable and it works! Instead of using concat() I used flatMap():

private void reloadDatabaseFromRest(final Place selectedPlace, final Event selectedEvent) {
    Completable.fromObservable(
            DatabaseManager.getInstance().clearDatabase()
                    .flatMap(aVoidDB -> mPlaceUseCase.getPlacesListFromRestObservable().filter(places -> places != null && !places.isEmpty()))
                    .flatMap(places -> mPlaceUseCase.savePlacesToRepository(places))
                    .flatMap(aVoidPlace -> mPlaceUseCase.saveSelectedPlace(selectedPlace))
                    .flatMap(aVoidSelectedPlace -> {
                        try {
                            return mEventUseCase.getEvenListObservableFromRest(selectedPlace.getId());
                        } catch (ParseException e) {
                            Log.e(TAG, e.getMessage());
                            return null;
                        }
                    })
                    .filter(eventList -> eventList != null)
                    .flatMap(eventList -> mEventUseCase.saveEventsToRepository(eventList))
                    .flatMap(aVoidEvents -> mEventUseCase.saveSelectedEventObservable(selectedEvent)))
            .subscribe(() -> {
                mSplashScreenUI.setupPlace(selectedPlace);
            });
}

Upvotes: 1

koperko
koperko

Reputation: 2487

You can use Completable type. Completable just informs subscriber that the stream is finished.

Wrap every observable you need to subscribe into a Completable with Completable.fromObservable() and then merge ( parallel start ) them with Completable.merge() or concat them ( in series one after the other ) with Completable.concat.

Note: make sure that all of the observables you wrap into Completable will call onComplete and so are not endless streams.

Upvotes: 4

Related Questions