RxJava объединить несколько наблюдаемых do sth после всех сделанных

Question

У меня есть метод, который должен перезагрузить БД с помощью нескольких наблюдаемых. В конце концов, он должен вызывать метод setupPlace (selecetedPlace), но только один раз. Каждый наблюдаемый возвращает другой объект (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)); 
} 

Но этот метод делает Подписаться() только на последнем наблюдаемым и другие не вызываются (я прав?). Я пробовал использовать фильтры, zip и делать глупые, если (o instanceOf ArrayList), но это неправильно. Может кто-то мне помочь, пожалуйста?

Answer 1

Вы можете использовать тип Completable. Завершенный только информирует абонента о завершении потока.

Wrap каждый наблюдаемым вам нужно подписаться на Completable с Completable.fromObservable(), а затем объединить (параллельный старт) их Completable.merge() или Concat их (последовательно, один за другим) с Completable.concat.

Примечание: убедитесь, что все наблюдаемое сматывание в Completable будет называть OnComplete и так не бесконечные потоков.

Answer 2

Большое спасибо @koperko! Он работает как шарм, и это именно то, что мне нужно. Во время метода перезаписи я заметил, что первый Observable испускает 6 элементов (clearDatabase()), поэтому я завернул его в Completable, и он работает! Вместо использования CONCAT() Я 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); 
      }); 
}