How to cancel an observable sequence

Question

I have a very simple IObservable<int> that acts as a pulse generator every 500ms:

var pulses = Observable.GenerateWithTime(0, i => true, i => i + 1, i => i,
                                         i => TimeSpan.FromMilliseconds(500))

And I have a CancellationTokenSource (that is used to cancel other work that is going on simultaneously).

How can I use the cancellation token source to cancel my observable sequence?

Answer 1

It is an old thread, but just for future reference, here is a simpler way to do it.

If you have a CancellationToken, you are probably already working with tasks. So, just convert it to a Task and let the framework do the binding:

using System.Reactive.Threading.Tasks;
...
var task = myObservable.ToTask(cancellationToken);

This will create an internal subscriber that will be disposed when the task is cancelled. This will do the trick in most cases because most observables only produce values if there are subscribers.

Now, if you have an actual observable that needs to be disposed for some reason (maybe a hot observable that is not important anymore if a parent task is cancelled), this can be achieved with a continuation:

disposableObservable.ToTask(cancellationToken).ContinueWith(t => {
    if (t.Status == TaskStatus.Canceled)
        disposableObservable.Dispose();
});