RxJS Observable：在处理最后一个订阅时执行清理？

Question

对于RxJS Observable的上次订阅处置时，执行副作用的最简洁方法是什么？在Observable已终止之前，这可能发生在。

假设我需要一个返回Observable的函数来发布对资源的更改。当所有订阅已被处置时，我想执行清理操作。

var observable = streamResourceChanges(resource); 
var subscription1 = observable.subscribe(observer1); 
var subscription2 = observable.subscribe(observer2); 
// ... 
subscription1.dispose(); // Does not perform the cleanup 
subscription2.dispose(); // Performs the cleanup 

我发现定义订阅处置行动的唯一方法是使用Rx.Observable.create。最后的处置可以通过共享订阅来处理，例如Observable.prototype.singleInstance()。

例如：

function streamResourceChanges(resource) { 
    return Rx.Observable.create(function(observer) { 
     // Subscribe the observer for resource changes... 
     // Return a cleanup function 
     return function() { 
      // Perform cleanup here... 
      console.log("Cleanup performed!"); 
     }; 
    }).singleInstance(); 
} 

有定义订阅处置副作用，类似于doOnNext，doOnCompleted或doOnError一个更合适的方法？

var withCleanup = withoutCleanup.doOnDispose(function() { 
    // Perform cleanup here... 
}); 

Answer 1

两个选择，浮现在脑海中，这取决于你的实际使用情况：

.finally()

source.finally(() => console.log("cleaning up")).singleInstance() 

.using()

Rx.Observable 
    .using(
     // allocate some disposable resource during subscribe. 
     // resource.dispose() will be called during unsubscribe. 
     () => new SomeResource(), 

     // use the disposable resource to create your observable 
     // for example... 
     resource => Rx.Observable.interval(resource.time)) 
    .singleInstance();