添加一些代码以清除问题
//generates a sequence in the range from input value (+1) to input value (+9)
Observable<ColoredIntegerModel> getSequenceObservable(int value, int delay, int color) {
return Observable.range(value+1,9)
.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);
});
}
//creates a stream if say input =2 of numbers from 1-20 (input*2) such that the output is 1 (Red color) 2-10(green color) 11 (Red color) 11-20 (Green Color)
Observable<ColoredIntegerModel> getEventStream(int value) {
return Observable.create(new ObservableOnSubscribe<ColoredIntegerModel>() {
@Override
public void subscribe(ObservableEmitter<ColoredIntegerModel> emitter) throws Exception {
for (int i = 0; i < value; ++i) {
ColoredIntegerModel model = new ColoredIntegerModel(i*10, Color.RED);
emitter.onNext(model);
Observable<ColoredIntegerModel> more = getSequenceObservable(i*10, 100, Color.GREEN);
more.subscribe(new Consumer<ColoredIntegerModel>() {
@Override
public void accept(ColoredIntegerModel coloredIntegerModel) throws Exception {
emitter.onNext(coloredIntegerModel);
}
});
}
}
});
}
上面的代码有效。它打印1(红色)2-10(绿色)11(红色),12-20,但我想要一个更清洁的解决方案。我也不确定何时可以处理getEventStream()中的内部订阅。
问题基本上是getEventStream为每个返回Observable的发射调用一个函数。这类似于Promise链,其中每个Promise都可以返回一系列其他Promise。希望这能澄清对原始问题的任何混淆。
如果您想要的是简化上面的代码,以便所有订阅处理留给最终订阅者并保持订单到后续排放,您可以这样做:
Observable<ColoredIntegerModel> getSequenceObservable(int value, int delay, int color) {
return Observable.range(value+1, 9)
.flatMap( i -> Observable
.just(new ColoredIntegerModel(i, color))
.delay(delay * (i + 1), TimeUnit.MILLISECONDS)
)
;
}
Observable<ColoredIntegerModel> getEventStream(int value) {
return Observable.range(0, value)
.concatMap(i ->
getSequenceObservable(i * 10,100, Color.GREEN)
.startWith(new ColoredIntegerModel(i*10, Color.RED))
)
;
}
如果你真的需要手动延迟,如果你不需要,只需将上面的getSequenceObservable
替换为:
Observable<ColoredIntegerModel> getSequenceObservable(int value, int color) {
return Observable.range(value+1, 9)
.map(i -> new ColoredIntegerModel(i, color))
;
}
你应该看看FlatMap operator
简而言之,它将Observable
中的每个元素转换为它自己的Observable
并加入它们。
对您的问题最简单的解决方案可能是:
getEventStream()
.flatMap(it -> getSequenceObservable(it))
.doOnNext(System.out::print)
.blockingSubscribe();
辅助函数的位置
static Observable<ColoredIntegerModel> getEventStream() {
return Observable.fromArray(
new ColoredIntegerModel(10, Color.RED),
new ColoredIntegerModel(20, Color.RED)
);
}
static Observable<ColoredIntegerModel> getSequenceObservable(ColoredIntegerModel color) {
return Observable.range(1, 10)
.flatMap(it -> Observable.timer(it, TimeUnit.SECONDS)
.map(time -> new ColoredIntegerModel(time.intValue(), Color.GREEN))
);
}
如果你想保持getEventStream()
的原始值你可以使用这样的东西而不是getSequenceObservable
static Observable<ColoredIntegerModel> getSequenceObservable(ColoredIntegerModel color) {
return Observable.range(1, 10)
.flatMap(it -> Observable.timer(it, TimeUnit.MILLISECONDS)
.map(time -> new ColoredIntegerModel(time.intValue(), Color.GREEN)))
.concatWith(Observable.just(color));
}
如果排放的排序很重要,请使用带有maxConcurrency的flatMap版本:
getEventStream()
.flatMap(it -> getSequenceObservable(it), true, 1)
.doOnNext(System.out::println)
.blockingSubscribe();