RXJava 在完成其他操作时创建一个 observable答案

【问题标题】：RXJava create an observable on completion of otherRXJava 在完成其他操作时创建一个 observable
【发布时间】：2018-09-05 20:37:04
【问题描述】：

嗨，我已经为此苦苦挣扎了一段时间。任何帮助表示赞赏。

我需要在完成另一个 observable 后运行一个 observable。所以例如下面的代码创建了一个从输入值到值 + 10 的 observable。

 Observable<ColoredIntegerModel> getSequenceObservable(int value, int delay, int color) {
    return Observable.range(value,10)
            .map(i -> {
                Log.d(TAG, "Value " + i
                        + " evaluating on " + Thread.currentThread().getName()
                        + " emitting item at " + System.currentTimeMillis());
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException e) {

                }
                return new ColoredIntegerModel(i, color);
            });
}

ColorIntegerModel如下

public class ColoredIntegerModel {

    private Integer mValue;
    private int mColor;

    public ColoredIntegerModel(Integer value, int color) {
        mValue = value;
        mColor = color;
    }

    public Integer getValue() {
        return mValue;
    }

    public int getColor() {
        return mColor;
    }
}

我如下创建两个 observables 并像这样连接它们。

Observable<ColoredIntegerModel> observable1 = getSequenceObservable(1, 1000, Color.BLUE);

Observable<ColoredIntegerModel> observable11 = getSequenceObservable(11, 1000, Color.RED);

 mDisposable =
            observable1.concatWith(observable11)
            .doOnDispose(() -> {Log.d(TAG, "observable disposed");})
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .repeat(2)
            .subscribe((m) -> {
                Utils.appendColoredText(mResultTextView, "Adding item "
                        + m.getValue().toString() + "\n", m.getColor());
            });

上面的代码打印出 1..10（蓝色，每个项目延迟 1s）和 11..20（红色）。

到目前为止一切顺利。

但我的要求是仅在第一个完成后创建第二个 observable。事实上，它可能是可观察对象的数组，其中 n+1 可观察对象仅在第 n 个完成后创建。每个 observable 可以发出多个项目。有没有运营商可以做到这一点？

【问题讨论】：

标签： android rxjs rx-java

【解决方案1】：

我不知道我是否理解正确，但如果您想在订阅时创建Observable，您需要defer 运算符

【讨论】：

谢谢我正在寻找延迟运营商。它间接解决了我的问题。

【解决方案2】：

你可以从前一个的 doOnCompleted() 开始下一个 observable

Observable<MyData> observable1 = ...;
Observable<MyData> observable2 = ...;
Observable<MyData> observable3 = ...;

Observable
       .concat(observable1.doOnCompleted(this::onCompleteObservable1),
            observable2.doOnCompleted(this::onCompleteObservable2),
            observable3.doOnCompleted(this::onCompleteObservable3))
            ...
            ...

希望对您有所帮助。

【讨论】：