最後のonNext（）実行のためのRxJavaコールバックが終了しました

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); 
       } 
      }); 
} 

問題：onNext(TempoAccount tempoAccount)からのすべての観測が完了する前にfilterAccountsWithProjects(accountsWithProjects)上記のコードでが発射されます。

編集：

私は、このようなタイプの観察可能作成したい：Observable<Map<TempoAccount, Collection<TempoProject>>を。 私は与えられた2つの観測があります。

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 Sを取得するには、そのアカウントのPairとTempoProjectのコレクションに各アカウントをマッピング。ペアをフィルタリングしてプロジェクトがあるかどうかを確認し、toMap()を使用して目的の結果を作成します。 toMap()を有効にするには、ストリームの終わりに達すると、オブザーバブルはonComplete()にコールする必要があることに注意してください。