javascript - 如何在 RxJS 中完成 Observable

Question

假设我们有一个Observable：

var observable = Rx.Observable
    .fromEvent(document.getElementById('emitter'), 'click');

我怎样才能使它完成（什么会触发所有订阅的观察者的onComplete事件）？

Answer 1

在这个现在的形式中，你不能。您的 observable 源自未完成的源，因此它本身无法完成。您可以做的是使用完成条件扩展此源。这将像：

var end$ = new Rx.Subject();
var observable = Rx.Observable
    .fromEvent(document.getElementById('emitter'), 'click')
    .takeUntil(end$);

当你想结束时observable，你就去做end$.onNext("anything you want here");。在这种情况下，结束事件是由您生成的。如果这是生成该事件的另一个源（按键等），那么您可以直接将从该源派生的可观察对象作为takeUntil.

文档：

• http://reactivex.io/documentation/operators/takeuntil.html
• https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/takeuntil.md

Answer 2

对我有用的是使用take()运算符。它将在 x 个事件后触发完整的回调。所以通过 1，它将在第一个事件之后完成。

打字稿：

private preloadImage(url: string): Observable<Event> {
    let img = new Image();
    let imageSource = Observable.fromEvent(img, "load");

    img.src = url;

    return imageSource.take(1);
}

Answer 3

我认为您正在寻找的是dispose()方法。

来自：https ://github.com/Reactive-Extensions/RxJS/blob/master/doc/gettingstarted/creating.md#cold-vs-hot-observables

请注意，subscribe 方法返回一个 Disposable，因此您可以取消订阅一个序列并轻松地处理它。当您在可观察序列上调用 dispose 方法时，观察者将停止侦听可观察数据。通常，您不需要显式调用 dispose，除非您需要提前取消订阅，或者当源 observable 序列的生命周期比观察者更长时。Rx 中的订阅是为不使用终结器的即发即弃场景而设计的。请注意，Observable 操作符的默认行为是尽快处理订阅（即，当发布 onCompleted 或 onError 消息时）。例如，代码将 x 订阅序列 a 和 b。如果 a 抛出错误，x 将立即从 b 取消订阅。

Answer 4

我为我的用例找到了一种更简单的方法，如果你想在 observable 完成时做一些事情，那么你可以使用这个：

const subscription$ = interval(1000).pipe(
  finalize(() => console.log("Do Something")),
).subscribe();

当所有订阅都取消订阅等时，会在完成时触发最终确定。