How to cancel asynchronous method called using Rx

I want to know how to cancel\stop asynchronous method called using reactive extensions using the TakeUntil.

So I want to be able to cance开发者_开发知识库l\stop the following method:

this.booksService.Search(searchText)
                 .ObserveOnDispatcher()
                 .Subscribe(results =>
                 {
                    this.books.Clear();
                    this.books.AddRange(results);
                 });

The return from Subscribe() is an IDisposable, when you dispose the value, like so:

var searchSub = this.booksService
                    .Search(searchText) ...
                    .Subscribe( ... );
// do something else
searchSub.Dispose();

the subscription is stopped. Does that actually stop the underlying booksSevice search? That depends on the implementation. If done w/ Observable.Create then it's quite possible to get that behavior.

You will have to tell booksService to cancel the search. The Search method starts the search and also returns an IObservable. When the search is complete OnNext will be called on the IObservable. There is some asynchronous code started in Search that you need to cancel.

My recommendation is to change the search method to an asynchronous call that takes a CancellationToken as the second argument. You can then convert this to an Observable sequence via Observable.ToAsync. Calling dispose on this observable will then trigger the cancellation of the CancellationToken. Doing this allows you to easily pass the token onto asynchronous web calls etc.

public Task Search(string searchText, CancellationToken cancel)
{
    // Pass the token on to the async webclient request etc.
}

var searchSub = Observable.ToAsync(c => this.booksService.Search(searchText, c))
                .Subscribe( ... );
// do something else
searchSub.Dispose();