RxJava callback для последнего выполнения onNext() завершено

Question

Я новичок в RxJava. Я хотел бы загрузить некоторые данные для каждого объекта TempoAccount из данной коллекции и сохранить все это на карте accountsWithProjects. Когда код последнего onNext(TempoAccount tempoAccount) будет завершен, я бы назвал метод filterAccountsWithProjects(accountsWithProjects). Есть ли простой способ добиться этого?

private void getProjectsForEachTempoAccount(Collection<TempoAccount> tempoAccounts) { 
    final Map<TempoAccount, Collection<TempoProject>> accountsWithProjects = 
      new HashMap<>(tempoAccounts.size()); 
    Observable<TempoAccount> accountsObservable = Observable.from(tempoAccounts); 
    accountsObservable 
      .compose(ObservableUtils.applySchedulers()) 
      .subscribe(new ObserverAdapter<TempoAccount>() { 
       @Override 
       public void onError(Throwable e) { 
        view.notifyAboutError(e.getMessage()); 
       } 

       @Override 
       public void onNext(TempoAccount tempoAccount) { 
        jira.requestProjectsInfoForTempoAccount(String.valueOf(tempoAccount.getId())) 
          .subscribeOn(Schedulers.io()) 
          .observeOn(AndroidSchedulers.mainThread()) 
          .subscribe(new ObserverAdapter<Collection<TempoProject>>() { 
           @Override 
           public void onError(Throwable e) { 
            view.notifyAboutError(e.getMessage()); 
           } 

           @Override 
           public void onNext(Collection<TempoProject> projects) { 
            accountsWithProjects.put(tempoAccount, projects); 
           } 

          }); 
       } 

       @Override 
       public void onCompleted() { 
        filterAccountsWithProjects(accountsWithProjects); 
       } 
      }); 
} 

Проблема: В приведенном выше коде filterAccountsWithProjects(accountsWithProjects) обжигают перед всеми наблюдаемыми из onNext(TempoAccount tempoAccount) завершены.

Edit:

Я хочу создать Наблюдаемые такого типа: Observable<Map<TempoAccount, Collection<TempoProject>>. У меня есть две наблюдаемые данные:

1. Observable<TempoAccount> accountsObservable = Observable.from(tempoAccounts)

2. Observable<Collection<TempoProject>> projectsForAccountObservable = jira.requestProjectsInfoForTempoAccount(TempoAccount account)

Так Мои вопросы: могу ли я connnect их как-то и создать карту, имеющую эти два наблюдаемыми.

Answer 1

Вы должны использовать функцию flatMap() в своем исходном потоке, чтобы делать то, что вы сейчас делаете в onNext(). Кроме того, вам не нужно фильтровать поток в onComplete(). Вы можете использовать filter() на самом потоке и решить проблему более «реактивным» способом.

Вот пример:

accountsObservable 
     .compose(ObservableUtils.applySchedulers()) 
     .map(tempoAccount -> new Pair<TempoAccount, Collection<TempoProject>>(tempoAccount, fetchInfoAccountForTempoAccount(tempoAccount))) 
     .filter(pair -> hasProjects(pair)) 
     .toMap(pair -> pair.first(), pair -> pair.second) 
     .subscribe(...) 

EDIT:

Обновленный предлагаемый ответ - вы получите TempoAccount с, то необходимо сопоставить каждую учетную запись на Pair учета и сбора TempoProject с , Вы фильтруете пары, чтобы увидеть, есть ли у вас какие-либо проекты, а затем вы используете toMap(), чтобы создать желаемый результат. Имейте в виду, что для работы toMap() ваши наблюдаемые должны вызвать onComplete(), когда достигнут конец потока.