rx-java - 为什么在此代码中未调用onCompleted？

我在viewmodel中有这个主题:

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();。

最佳答案

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


PublishSubject以及其他*Subjects将一直存在，直到有人调用onError或onCompleted为止。实际上，它们在这方面与任何其他 Observable 都没有什么不同。
newArticleSubject创建新的PublishSubject，并将发射的每个项目平面映射到不同的可观察对象。如果getArticleById以错误结束，则flatMap返回的observable也将结束。但是，当getArticleById发出并正常完成时，flatMap将继续等待articleSubject发出新项目。如果您需要明确结束articleSubject，只需致电articleSubject.onCompleted()

关于rx-java - 为什么在此代码中未调用onCompleted？，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/40222803/