RxJava Single.concat

Question

我有以下代码：

 model.getCategories()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<Category>>()
                {
                    @Override
                    public void call(final List<Category> categories)
                    {
                        model.getUnits()
                                .subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Action1<List<Unit>>()
                                {
                                    @Override
                                    public void call(List<Unit> units)
                                    {
                                        view.showAddProductDialog(units, categories);
                                    }
                                });
                    }
                });

我有这个丑陋的嵌套。我该如何解决它。我试过这样的事情：

Single.concat(model.getCategories(), model.getUnits())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<? extends Object>>()
                {
                    @Override
                    public void call(List<? extends Object> objects)
                    {
                        // do something
                    }
                });

但问题是我无法确定是List<Category>还是List<Unit>来了。

有没有一种方法可以使用 concat 并检测来自哪种流（List<Category> 或 List<Unit> 或其他）？

我还需要检测所有观察者是否已完成执行另一个操作。

谢谢指教。

Answer 1

使用Single.zip():

Single.zip(
    getCategories().subscribeOn(Schedulers.io()),
    getUnits().subscribeOn(Schedulers.io()),
    (a, b) -> Pair.of(a, b)
)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
    pair -> view.showAddProductDialog(pair.first, pair.second),
    error -> showError(error.toString())
)

其中 Pair 可以是任何元组实现，即 this.

RxJava Single.concat

RxJava Single.concat

rx-java2