使用RxJava将多个项目从可观察到的对象组合到新对象

Question

我正在尝试实现某种流式解析器。假设我有整数的stream，我将它们组合起来创建新的object，它聚合了stream的一部分。

例如，当integer为负数时，object为“done”。为了保持简单，生成的项目将是一串数字。

下面是一个简单的例子：

Source: 1, -2, 3, 4, -5, 6, -7, 8, 9, 10, -11
Output: "1-2", "343-5", "6-7", "8910-11"

rx.Observable<Integer> src = rx.Observable
    .from(Arrays.asList(1, -2, 3, 4, -5, 6, -7, 8, 9, 10, -11));

rx.Observable<String> res = src
    .lift(new rx.Observable.Operator<String, Integer> () {

        @Override
        public rx.Subscriber<? super Integer> call(
            final rx.Subscriber<? super String> subscriber) {

            return new rx.Subscriber<Integer>(subscriber) {

                private final StringBuilder cur = new StringBuilder();

                @Override
                public void onCompleted() {
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onCompleted();
                    }
                }

                @Override
                public void onError(Throwable e) {
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onError(e);
                    }
                }

                @Override
                public void onNext(Integer i) {
                    if (subscriber.isUnsubscribed())
                        return;

                    cur.append(i).append('');
                    if (i < 0) {
                        subscriber.onNext(cur.toString());
                        cur.delete(0, cur.length());
                    }
                }
            };
        }
    });

rx.observers.TestSubscriber<String> sub =
    new rx.observers.TestSubscriber<>();
res.subscribe(sub);
List<String> l1 = sub.getOnNextEvents();

rx.Observable<String> res2 = res.take(2);

看起来每个输入onNext都应该调用output onNext。

有什么想法吗？

Answer 1

嗯。写长篇文章确实有助于带来新的想法，而且很管用。（我撞上墙已经两天了）。

生成observer 并发出observer.empty()直到对象就绪。flatmap之后。

下面是工作示例：

rx.Observable<rx.Observable<String>> res = src
    .lift(new rx.Observable.Operator<rx.Observable<String>, Integer> () {

        @Override
        public rx.Subscriber<? super Integer> call(
            final rx.Subscriber<? super rx.Observable<String>> subscriber) {

            return new rx.Subscriber<Integer>(subscriber) {

                private final StringBuilder cur = new StringBuilder();

                // onError, onCompleted skipped

                @Override
                public void onNext(Integer i) {
                    rx.Observable<String> out;
                    cur.append(i);
                    if (i < 0) {
                        out = rx.Observable.just(cur.toString());
                        cur.delete(0, cur.length());
                    } else {
                        out = rx.Observable.<String>empty();
                    }
                    subscriber.onNext(out);
                }
            };
        }
    });

rx.Observable<String> res2 = res
    .flatMap(stringObservable -> stringObservable)
    .take(2);