用RxJS处理流端的惯用方法

Question

我需要在流结束时做一些操作。习惯的做法是什么?

现在我使用下面的代码:

source.subscribe(undefined, undefined, function() {
  socket.send({type: 'end'});
});

Answer 1

有几种方法可以做到这一点:

1. 使用操作符subscribeOnCompleted()而不是将空值传递给.subscribe方法。

2. 使用tapOnCompleted()，如上所述，但它不会启动序列，您可以通过序列注入它的一部分。

3. 使用.finally()，它将在序列完成时执行(正常或不正常)。

4. 在你的例子中，你显示了一个副作用，但如果你正在清理资源，使用.using()将更加语义化，它接受一次性并将其绑定到订阅的生命周期。

顺序如下:

1. source.subscribeOnCompleted(() => socket.send({type: 'end'}));

2. source.tapOnCompleted(() => socket.send({type: 'end'})).subscribe()

3. source.finally(() => socket.send({type: 'end'})).subscribe()

4. Rx.Observable.using(() => createResource(), (resource) => source).subscribe()