最后一个onNext（）执行完成的RxJava回调

Question

我是RxJava新手。我想从给定集合中下载每个TempoAccount实体的一些数据，并将其存储在map 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）在完成下一个（TempoAccount TempoAccount）中的所有观察之前被激发。

编辑：

我想创建这样一个类型的可观察对象：<代码>可观察

>

可观察

<代码>可观察

所以我的问题是：我能以某种方式将它们连接起来并创建具有这两个可观察对象的地图吗？




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(...)

编辑：

更新了建议的答案-您获得了TempoAccount，然后将每个帐户映射到帐户和TempoProject集合的一对。您过滤这些对以查看是否有任何项目，然后使用toMap（）创建所需的结果。请注意，要使toMap（）工作，您的可观察对象必须在到达流的末尾时调用onComplete（）。