Cancellation Documentation

[aldendaniels]

The RxJS docs state that:

Cancellation, exceptions, and synchronization are also handled gracefully by using the methods on the Observable object.

I'm struggling to find documentation regarding cancellation. Specifically, I'm looking for a way to propagate a cancelation event backwards through the observable chain to the observable source.

For example, imagine that I'm aggregating the result of three XHR requests. If one of the requests responds with an error code, I want to cancel the other two outgoing requests.

Does RxJS provide an idiom for this kind of cancellation? If not, it would help to clarify what is meant by "cancellation" in the docs.

Thanks!

[gre]

I would be also interested to know about this.

Also, in this example: https://github.com/Reactive-Extensions/RxJS/blob/master/examples/autocomplete/autocomplete.js ,
the Autocomplete does not seem to "cancel" the previous request (let's imagine network is slower than then thresholding set out there, there will be a lot of pending requests).

Is there a simple way we could refactor this example to fix this in a RxJS style?

Thanks

[paulpdaniels]

Cancellation comes in two flavors: explicit and implicit.

Explicit

This refers to actively disposing of a subscription when you are done using it.

var subscription = Rx.Observable
   .interval(2000).scan(0, (acc, x) => acc + x)
   .subscribe(total => console.log(total));

//Some time in the future
subscription.dispose();

In the above, I am cancelling the subscription to the observable explicitly which immediately cancels the future scheduled actions.

Implicit

This is more toward @gre 's comment.

Probably the more common scenario, here the subscription is cancelled internally by RxJS based on some condition being met.

To take the example of the autocomplete

    var searcher = keyup.flatMapLatest(searchWikipedia);

First thing I would recommend is that you go look at the documentation for flatMapLatest. In a nutshell it will emit only from the most recent sequence, while cancelling the previously held one. So yes, it does cancel the previous request just as soon as the next request comes in.

Now what that cancellation means really depends on the sequence and what the underlying source supports. RxJS always provides best effort cancellation for in-flight requests. However this won't always translate to cancelling all the in motion actions, the only guarantee you have is that your down stream Observers will no longer receive events from cancelled subscriptions.

If you take the example of the autocomplete, since the ajax request is a promise (which cannot be cancelled) cancelling it will only stop the event from being emitted not the ajax request. Something like the FileReader API which has abort could be cancelled in flight but again it will all come down to what the underlying api supports.

[gre]

Actually an XMLHttpRequest does have an .abort() method. So I think we need
a better paradigm than promise in this example.
On Jul 20, 2015 4:03 AM, "Paul Daniels" notifications@github.com wrote:

Cancellation comes in two flavors: explicit and implicit.

Explicit

This refers to actively disposing of a subscription when you are done
using it.

var subscription = Rx.Observable
.interval(2000).scan(0, (acc, x) => acc + x)
.subscribe(total => console.log(total));
//Some time in the future
subscription.dispose();

In the above, I am cancelling the subscription to the observable
explicitly which immediately cancels the future scheduled actions.

Implicit

This is more toward @gre https://github.com/gre 's comment.

Probably the more common scenario, here the subscription is cancelled
internally by RxJS based on some condition being met.

To take the example of the autocomplete

var searcher = keyup.flatMapLatest(searchWikipedia);

First thing I would recommend is that you go look at the documentation for
flatMapLatest
https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/flatmaplatest.md.
In a nutshell it will emit only from the most recent sequence, while
cancelling the previously held one. So yes, it does cancel the previous
request just as soon as the next request comes in.

Now what that cancellation means really depends on the sequence and what
the underlying source supports. RxJS always provides best effort
cancellation for in-flight requests. However this won't always translate to
cancelling all the in motion actions, the only guarantee you have is that
your down stream Observers will no longer receive events from cancelled
subscriptions.

If you take the example of the autocomplete, since the ajax request is a
promise (which cannot be cancelled) cancelling it will only stop the event
from being emitted not the ajax request. Something like the FileReader
API which has abort could be cancelled in flight but again it will all
come down to what the underlying api supports.

—
Reply to this email directly or view it on GitHub
#817 (comment)
.

[paulpdaniels]

I can't speak for the original intent of the example, it may be trying to highlight how to strangle out old code, or how RxJS plays nice with other libraries. That being said, it might make sense to use RxJS-DOM which does have abortable Ajax. But it would need to make that inclusion clear.

[gre]

I guess that is already possible no? https://github.com/Reactive-Extensions/RxJS-DOM/blob/master/src/ajax.js#L186

I don't know the syntax however.
Sorry to drift from original post ;)

[mattpodwysocki]

@gre yes, it is possible to use rx.dom.js, but in this simple case it really doesn't matter all that much as I'm showing how well it works with other libraries. Surely if you want to have cancellation, then go ahead and use rx.dom.js.

[mattpodwysocki]

@aldendaniels I think @paulpdaniels answered this correctly as it is a best try cancellation which means if the XHR/Promise/Whatever has not completed, then it will call abort if the Observable is created with a Disposable which destroys or aborts the resource. If, however, it has already completed, then the value from the XHR/Promise/Whatever is simply ignored.

[gre]

@mattpodwysocki : thanks for the feedback, great to know we can do it, I was just very curious how easy it was to make it :)

BTW in RxJS-DOM, the autocomplete example uses jsonp, that is cancellable, so great :)
https://github.com/Reactive-Extensions/RxJS-DOM/blob/master/examples/Autocomplete/autocomplete.js

Cheers

[aldendaniels]

@mattpodwysocki, @paulpdaniels - Thanks!

I'm new to RX, but I've used Promises extensively. The promise spec doesn't (yet) provide cancellation semantics, but some promise libraries are inventing there own.

Since a Promise only returns a single value, it has a "pending" state while the promise is awaiting resolution. Since the promise has a pending state, it makes sense that you can cancel a promise when (and only when) it's pending.

Cancellation implies two things:

1. An onCancel handler (if defined) is called on upstream promises that don't have any other subscribers.
2. The memory associated with those upstream promise chains will be freed since no value is now expected.

Check out petkaantonov/bluebird#415 (comment) for more on this.

Example:

let dataPromise = fetch(ur)
  .then(transformData); 

window.setTimeout(() => {
  dataPromise.cancel(); // Calls some onCancel() method that fetch() setup.
}, 100);

This is great for Promises, but doesn't map easily to observables. With observables, we can shut down the pipe (unsubscribe), but not cancel the in-flight operation.

From what I can tell, Rx observable subscribers don't know when an observable is "pending". If you had a way to tell an observable chain to expect a currently pending value, then it might be possible to cancel the currently pending action.

You could probably accomplish this today via an observable stream of promises. If you're mapping events to API requests for example, this would become a sync operation that maps events to promises. Then those promises could be cancelled using something like Bluebirds upcoming cancellation API.

Am I thinking correctly?

[paulpdaniels]

@aldendaniels As was mentioned above, Rx uses best effort cancellation, meaning that it will do whatever it reasonably can to clean up the resources it created. When dealing with Promises that means simply ignoring the value (it'd be great to use something like bluebirds cancellation, but since the actual specification doesn't include cancellation, it isn't really an option). In most cases though we can actually cancel the in-flight operation.

It is important to note as well that explicit cancellation is not part of the Observable as it is with promises. The Disposable returned when an Observer subscribes to an Observable is what contains the logic for cancellation. This is why when creating new operators with Observable.create there should generally be a Disposable or function returned that when called would dispose of any resources that had been allocated.

This kind of cancellation covers not just things like removing event handlers, but in the case of certain operators like controlled and interval, it will also deallocate buffered events and cancel any pending operations. In the latter operator it can do this because Scheduler also returns a Disposable one of the schedule overloads is called. This Disposable allows pending (or in-flight) events to be cancelled as well.

Does that make it any clearer? It also might be something to bring up on the gitter to feedback faster.

[aldendaniels]

@paulpdaniels - Was clearing out old emails and realized that I never responded to your post. My apologies! This is very helpful and a great starting point for getting my mind around cancellation with RxJS. Thanks for taking the time to help explain.