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