Cancel a task in reactivex.net

Question

Assumed that I have existed code like:

public IEnumerable<DataType> GetAllData(string[] ids) {
    foreach(var id in ids) {
        //this is a time-consuming operation, like query from database
        var data = this.repo.Find(id);
        yield return data;
    }
}

I tried to apply Rx to the front-end code:

var observable = GetAllData(new[] { "1", "2", "3" }).ToObservable();
var subs = observable
    .SubscribeOn(Scheduler.Default)
    .Subscribe(
        data => Console.WriteLine(data.Id), 
        () => Console.WriteLine("All Data Fetched Completed"));

And it's working properly.

But once I bind a subscription to the IObservable, is there any way I can stop it continue fetching data half-way? Dispose the subscription won't stop the enumeration.

score 0 · Answer 1 · edited May 23 '17 at 12:23

Well, a simple approach is:

var cts = new CancellationTokenSource();
var observable = GetAllData(new[] { "1", "2", "3" }).ToObservable().TakeWhile(x => !cts.IsCancellationRequested);
var subs = observable
    .SubscribeOn(Scheduler.Default)
    .Subscribe(
        data => Console.WriteLine(data.Id), 
        () => Console.WriteLine("All Data Fetched Completed"));
//...
cts.Cancel();

https://stackoverflow.com/a/31529841/2130786

Cancel a task in reactivex.net

1 Answers1