rx-java - 在RxJava中，我无法从flatMap发出onComplete

我有一个问题，当我使用平面图时，我的可观察链不会终止。

我把我的例子归结为：

int count = Observable.just(1,2,3)
        .flatMap(s -> Observable.<Integer>create(subscr-> {
            subscr.onNext(s);
            if(s>2) {
                subscr.onCompleted();
            }
        }))
        .doOnEach(a->System.err.println(a.getKind()+" -> "+a.toString()))
        .count()
        .toBlocking()
        .first();
    System.err.println("Count: "+count);

我希望'doOnEach'报告三个onNext事件，然后报告一个onCompleted，最后该链应该终止。

但是，输出是这样的：

OnNext -> [rx.Notification@c9e6d24d OnNext 1]
OnNext -> [rx.Notification@c9e6d24e OnNext 2]
OnNext -> [rx.Notification@c9e6d24f OnNext 3]

（然后继续挂起）

如果删除flatMap运算符：

int count = Observable.just(1,2,3)
        .doOnEach(a->System.err.println(a.getKind()+" -> "+a.toString()))
        .count()
        .toBlocking()
        .first();
System.err.println("Count: "+count);

...完全符合预期：

CREATED!
OnNext -> [rx.Notification@e3598bd9 OnNext 1]
OnNext -> [rx.Notification@e3598bda OnNext 2]
OnNext -> [rx.Notification@e3598bdb OnNext 3]
OnCompleted -> [rx.Notification@3834d63f OnCompleted]
Count: 3

我想我做错了什么（我无法想象在这种基本情况下会出现错误），但是我看不到它。

感谢您的任何帮助...（我正在使用RxJava 1.1.0）

最佳答案

问题在以下几行中：

        if(s>2) {
            subscr.onCompleted();
        }

Observable和s == 1的s == 2个不会发出onCompleted。由于flatMap仅在所有onCompleted完成后才发出Observable，因此就您而言，它不会发出onCompleted。