CODEWITHSUNDEEP
c#system.reactive
Cristian Diaconescu
Cristian Diaconescu

Reputation: 35631

Rx - How to create IObservable<T> from Task<T> such that unsubscribing cancels the task?

I'm new to Rx so bear with me.

I want to wrap a Task<T> in an IObservable<T>. So far so good: 

Task<T> task = Task.Factory.StartNew(...);
IObservable<T> obs = task.ToObservable();

Now, what I want is to signal the task to cancel when the observer unsubscribes: 

var cancel = new CancellationToken();
Task<T> task = Task.Factory.StartNew(..., cancel);

IObservable<T> obs = task.ToObservable(); //there should be a way to tie the cancel token
                                          //to the IObservable (?)

IDisposable disposable = obs.Subscribe(...);
Thread.Sleep(1000);
disposable.Dispose(); // this should signal the task to cancel

How do I do that?

FWIW here's the scenario that generated this tangent: Rx and tasks - cancel running task when new task is spawned?

Upvotes: 2

Views: 1948

Answers (2)

Timothy Shields
Timothy Shields

Reputation: 79441

Suppose you have a method like this:

Task<Gizmo> GetGizmoAsync(int id, CancellationToken cancellationToken);

You can turn that into an IObservable<Gizmo> where subscribing starts the Task<Gizmo> and unsubscribing cancels it by using the following.

IObservable<Gizmo> observable = Observable.FromAsync(
    cancellationToken => GetGizmoAsync(7, cancellationToken));

// Starts the task:
IDisposable subscription = observable.Subscribe(...);

// Cancels the task if it is still running:
subscription.Dispose();

Upvotes: 2

Cory Nelson
Cory Nelson

Reputation: 29981

Here's the simplest way I can think of, using Observable.Create:

static IObservable<int> SomeRxWork()
{
    return Observable.Create<int>(o =>
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        IDisposable sub = SomeAsyncWork(cts.Token).ToObservable().Subscribe(o);
        return new CompositeDisposable(sub, new CancellationDisposable(cts));
    });
}

static Task<int> SomeAsyncWork(CancellationToken token);

The initial way I hinted at in the comments is actually rather verbose:

static IObservable<int> SomeRxWork()
{
    return Observable.Create<int>(async (o, token) =>
    {
        try
        {
            o.OnNext(await SomeAsyncWork(token));
            o.OnCompleted();
        }
        catch (OperationCanceledException)
        {
        }
        catch (Exception ex)
        {
            o.OnError(ex);
        }
    });
}

Upvotes: 3

Related Questions