RxJS Observed: Does it clean when the last subscription is deleted?

Question

What is the cleanest way to perform a side effect when posting the latest subscription to an RxJS Observable? This can happen before the Observable completes .

Let's say I need a function that returns an Observable that emits resource changes. I would like to perform a cleanup operation when all subscriptions have been deleted.

 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 

The only thing I found to determine the disposition recycling action is to use Rx.Observable.create . Last deletion can be controlled by sharing a subscription, such as Observable.prototype.singleInstance() .

For instance:

 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(); } 

Is there an easier way to define a side effect for deleting a subscription similar to doOnNext , doOnCompleted or doOnError ?

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

Answer 1

Depending on your actual use case, two options arise:

. 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();