使用 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()