rx-java - 为什么在这段代码中没有调用 onCompleted？

Question

我在 View 模型中有这个主题:

private PublishSubject<String> articleSubject;

public Observable<Article> newArticleSubject() {
    articleSubject = PublishSubject.create();
    return articleSubject.flatMap(new Func1<String, Observable<Article>>() {         
        @Override
        public Observable<Article> call(String articleId) {
            return dataModel.getArticleById(articleId).subscribeOn(Schedulers.newThread());
        }
    });
}

我像这样绑定(bind) View :

viewModel.newArticleSubject()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new ArticleSubscriber());

这是我的订阅者:

 private class ArticleSubscriber extends Subscriber<Article> {

        @Override
        public void onCompleted() {
            Log.d("test", "This is not ever printed");
        }

        @Override
        public void onError(Throwable e) {
            //do stuff
        }

        @Override
        public void onNext(Article article) {
            //do stuff
        }
}

onCompleted从未调用过，但 onError和 onNext是。为什么？内部 dataModel.getArticleById(articleId) , subscriber.onCompleted();叫做。

Answer 1

从 PublishSubject the documentation 中可用的图像可以得出结论。 :


PublishSubject以及其他*Subjects会一直活到有人来电 onError或 onCompleted .事实上他们在这件事上没有什么不同来自任何其他 Observable .
newArticleSubject创建新 PublishSubject flat 将发射的每个项目映射到不同的 observable。万一getArticleById以 flatMap 返回的 observable 错误结束也结束了。然而当 getArticleById正常发出并完成 flatMap将继续等待 articleSubject 发出新项目.如果您需要明确结束 articleSubject只需调用articleSubject.onCompleted()