Why RxJS function toPromise is not unsubscribing from subscription

Question

In our app we have quite a few places where we use:

someObservable.take(1).subscribe(onSuccessHandler, onFailureHandler);

But with subscriptions you need to worry about unsubscribing at some point and that is not always straightforward.

I was thinking of simplifying this and rewriting this as:

someObservable.toPromise().then(onSuccessHandler).catch(onFailureHandler);

But looking at the implementation of toPromise() (here) I don't seem to understand why it doesn't care about unsubscribing.

The comment in the code says that no cancellation could be done, but how we just leave it like this to leak memory (in case we actually are).

EDIT

I came up with an example that worries me:

Observable.timer(10, 10).toPromise().then((v) => console.log("I'm done"));

If observables I retrieve are such that they never complete, then not only my promises never complete with a value, but also I have no way to unsubscribe from such observables (e.g. time-out them and my promises), because I have no access to a subscription object. And that does leak memory!

Answer 1

My guess is that it is because a promise is resolved only once (not like a sequence which is a stream). Note that the subscriber always keep the latest value, rejects on failure and resolves to the latest value on complete.

To try it for your self, try:

    Observable.timer(300,300).take(4).toPromise().then((v) => console.log('tick: ', v)); // logs "3", the last element
    Observable.from(['a','b','c']).toPromise().then((v) => console.log('tick: ', v)); // logs "c", the last element

As for unsubscribing, it is automatic on completion, so before completion, you shouldn't unsubscribe, on completion, you resolve the promise and silently (due to the default behavior of observables) unsubscribed.

Answer 2

'No cancellation can be done' comment probably refers to cancellation of the promise which is something entirely different from observable completion. About the implementation of that operator, you will see if you observe carefully, that the .subscribe is called with its three parameters :

• onNext handler which holds the streamed value
• onError handler which rejects the promise
• completion handler which resolve the promise with the held value

Then Rxjs streams works so that when a stream is completed there is an automatic chain of unsubscriptions taking place, the same way that when a stream is subscribed to, there is an automatic chain of subscriptions taking place.

For more infos have a look here