RxJava 基于回调向流添加事件

Question

添加一些代码来解决问题

//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(Red) 2-10(Green) 11(Red), 12-20，但我想要一个更干净的解决方案。我也不确定何时可以处理 getEventStream() 中的内部订阅。

问题基本上是 getEventStream 为每个发射调用一个函数，该函数也返回一个 Observable 。这类似于一个 Promise 链，其中每个单独的 Promise 可以返回一系列其他 Promise。希望这可以澄清对原始问题的任何混淆。