如何在 RxJS 中完成 Observable

Question

假设我们有一个 Observable:

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

我怎样才能做到 Complete（什么会触发所有订阅的 Observers[=19= 的 onComplete 事件]) ?

Answer 1

以目前的形式，你不能。您的可观察对象源自未完成的来源，因此它本身无法完成。您可以做的是使用完成条件扩展此源。这会像：

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

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

的参数

文档：

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

Notice that the subscribe method returns a Disposable, so that you can unsubscribe to a sequence and dispose of it easily. When you invoke the dispose method on the observable sequence, the observer will stop listening to the observable for data. Normally, you do not need to explicitly call dispose unless you need to unsubscribe early, or when the source observable sequence has a longer life span than the observer. Subscriptions in Rx are designed for fire-and-forget scenarios without the usage of a finalizer. Note that the default behavior of the Observable operators is to dispose of the subscription as soon as possible (i.e, when an onCompleted or onError messages is published). For example, the code will subscribe x to both sequences a and b. If a throws an error, x will immediately be unsubscribed from b.

Answer 4

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

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

finalize 在完成时触发，当所有订阅都取消订阅等时

如何在 RxJS 中完成 Observable

How can I complete Observable in RxJS

javascript

stream

observable

rxjs

reactivex