onComplete 在 RxJS 中是如何工作的

Question

如果 Observables 是从有限数据构建的，它们自然会完成。

import {Observable, Subject} from "rx";

let stream0$ = Observable.of("1", "2", "3");
let stream1$ = stream0$.map(x => x);

stream1$.subscribe(
  (val) => { console.log("onNext", val) },
  (err) => { console.log("onError", err) },
  ()    => { console.log("onCompleted") }
);

// onNext 1
// onNext 2
// onNext 3
// onCompleted

或者如果没有就不要。但是订阅主题的可观察对象呢？例如：

import {Observable, Subject} from "rx";

let subj$ = new Subject();
let stream1$ = subj$.map(x => x);

stream1$.subscribe(
  (val) => { console.log("onNext", val) },
  (err) => { console.log("onError", err) },
  ()    => { console.log("onCompleted") }
);

subj$.onNext("foo");

// onNext foo

"onCompleted" 虽然源已结束 [=21=]，但未记录。我们能否以某种方式将此 "end" 事件传递给 stream1$。我在文档中没有找到关于这个重要内容的信息。很高兴看到像这里这样的图来确定事件流。

Answer 1

有了主题，你就完全掌控了。 Rx.Subject 实现观察者接口，调用 onNext.

时使用的是 observer interface

一门学科

assume that all serialization and grammatical correctness are handled by the caller of the subject.

这意味着由您来指示完成和错误。要发出完成信号，请使用 onCompleted。仅供参考，这是上述语法：

This grammar allows observable sequences to send any amount (0 or more) of onNext messages to the subscribed observer instance, optionally followed by a single success (onCompleted) or failure (onError) message.

The single message indicating that an observable sequence has finished ensures that consumers of the observable sequence can deterministically establish that it is safe to perform cleanup operations.

A single failure further ensures that abort semantics can be maintained for operators that work on multiple observable sequences.

注意：对于 RxJS v5，观察者界面已更改，请参见。 new interface

Answer 2

How Complete and Error events actually work in RxJS 是我所做的后续研究。

引用我自己的话。

Completion should not be taken as "event at the end of the program" or something. It's a very specific thing. There are only three possible ways for stream to complete: 1) Be a finite Observable and complete naturally 2) Be a Subject and get imperative onCompleted() call. 3) Get completion event from upstream. Any form of process termination / unsubscription do not complete streams.

Completion terminates stream. Nothing happens in a stream after completion.

Completion is passed downstream. Observable derived from Subject completes if/when Subject completes. Subject subscribed to Observable completes if/when Observable completes.

onComplete 在 RxJS 中是如何工作的

How onComplete actually works in RxJS

javascript

rxjs