RxJava: функция вызова при завершении двух наблюдаемых с различными типами результатов

Question

Я работаю над приложением Android, которое извлекает два разных объекта через API, к которому я обращаюсь через RxJava Observables. Чтобы обновить пользовательский интерфейс, мне нужны и результаты.

Как я могу запустить функцию, как только оба наблюдателя завершатся? Кажется, что такие функции, как merge, делают то, что я планирую, но насколько я вижу, они работают только для Observables с одинаковым типом результата или для него нужен составной объект, который может представлять оба типа.

Простой пример:

Observable.just("Hello world") 
     .subscribe(new Action1<String>() { 
      @Override 
      public void call(String s) { 
       System.out.println(s); 
      } 
     }); 
Observable.just(1, 2, 3, 4, 5, 6) 
     .subscribe(new Action1<Integer>() { 
      @Override 
      public void call(Integer i) { 
       System.out.println(i); 
      } 
     }); 

Что мне нужно сделать, чтобы запустить System.out.println("Finished!") как только оба Наблюдаемые выполнили свою задачу?

В конкретном случае моего приложения для Android я мог просто сохранить результаты в реальном классе, иметь функцию updateUi, которая работает только тогда, когда все необходимые данные уже появились и вызывают эту функцию как из onCompleted звонков, но я чувствую Существует лучший способ.

Answer 1

Версия 2: испускаемые элементы обрабатываются как «боковые» действия, только событие onCompleted переходит к объединенному наблюдаемому.

Observable<String> stringObservable = Observable.just("Hello world") 
     .doOnNext(System.out::println) 
     .ignoreElements(); 
Observable<Integer> integerObservable = Observable.just(1, 2, 3, 4, 5, 6) 
     .doOnNext(System.out::println) 
     .ignoreElements(); 

Observable.merge(stringObservable, integerObservable) 
     .subscribe(new Subscriber<Object>() { 
      @Override 
      public void onCompleted() { 
       System.out.println("Finished!"); 
      } 

      @Override 
      public void onError(Throwable throwable) { 
       /*optionally handle if one of the observables calls onError()*/ 
      } 

      @Override 
      public void onNext(Object o) { /*not called*/ } 
     }); 

Answer 2

Думаю, вам нужен TaskCoordinator. Храните простой счетчик в taskCoordinator. Каждый раз, когда вы сигнализируете об успехе, счетчик будет уменьшаться. Когда счетчик достигнет нуля, обратный вызов будет выполнен. Вы можете увидеть мою реализацию следующим образом: https://github.com/codefacts/crm-common/blob/master/src/main/java/io/crm/util/TaskCoordinator.java

Это лучше, потому что каждый раз, когда вы координируете задания, это пригодится. Реактивный поток не имеет поддержки для координации задач. Таким образом, в этом случае будет полезно реализовать библиотеку обещаний или реализацию задачи.