Cancelling / Ignoring result of ReactiveAsyncCommand in ReactiveUI

Question

I have a ReactiveAsyncCommand to perform a search as a user is typing into a textbox, which is setup like this:

var results = SearchCommand.RegisterAsyncFunction(term => 
PerformSearch((string)term));

this.ObservableForProperty(x => x.SearchTerm)
.Throttle(TimeSpan.FromMilliseconds(800))
.Select(x => x.Value).DistinctUntilChanged()
.Where(x => !String.IsNullOrWhiteSpace(x))
.InvokeCommand(SearchCommand);

_SearchResults = results.ToProperty(this, x => x.SearchResults);

The trouble is that the search function can be quite slow due to the database query it needs to perform and an outdated result is displayed, which I believe is due to ReactiveAsyncCommand not running again until the current async task is completed.

So my question is, how can I either cancel the async task that is running and start again with the current search term or drop the result entirely if it is not for the current search term.

It seems the same as the second part of this discussion, but I'm not sure how to apply that to my code as my search code returns an IEnumerable not an IObservable.

Note this for RxUI 4 as it's a .NET 4 application.

Update: PerformSearch method

private List<WizardLocationSearchResult> PerformSearch(string searchTerm)
{
var results = new List<WizardLocationSearchResult>();
bool isMatch = false;


if (Regex.IsMatch(searchTerm, _postcodeRegex, RegexOptions.IgnoreCase))
{
    var locationResult = _locationService.GetByPostcode(searchTerm);
    _locationService.DeepLoad(locationResult, true, Data.DeepLoadType.IncludeChildren, typeof(TList<EnterpriseAndHolding>));

    results.AddRange(ProcessLocationSearches(locationResult));
    isMatch = true;
}


if (!isMatch)
{
    var query = new LocationParameterBuilder(true, false);
    string formattedSearchTerm = searchTerm + "%";
    query.AppendLike(LocationColumn.Address1, formattedSearchTerm);
    query.AppendLike(LocationColumn.Address2, formattedSearchTerm);
    query.AppendLike(LocationColumn.Town, formattedSearchTerm);
    query.AppendLike(LocationColumn.PostalTown, formattedSearchTerm);
    query.AppendLike(LocationColumn.County, formattedSearchTerm);

    var locationResult = _locationService.Find(query.GetParameters());
    _locationService.DeepLoad(locationResult, true, Data.DeepLoadType.IncludeChildren, typeof(TList<EnterpriseAndHolding>));
    results.AddRange(ProcessLocationSearches(locationResult));
}

return results;
}

Answer 1

The idea behind ReactiveAsyncCommand is that it intentionally constrains the number of in-flight requests. In this case, you want to forget about that constraint, so let's use regular ReactiveCommand instead:

SearchCommand
    .Select(x => Observable.Start(() => PerformSearch(x), RxApp.TaskPoolScheduler))
    .Switch()
    .ToProperty(this, x => x.SearchResults);

Note that Select.Switch here is similar to SelectMany, except that it will always maintain the order of inputs while throwing away old inputs if they don't complete in time.

Blah blah CancellationTokenSource blah blah TPL

In this case, the underlying method itself (_locationService.DeepLoad) is synchronous - it's not possible to cancel the method in a safe way, so we have to just let cancelled results run to completion and ignore the result, which is what we'll end up doing here.

Edit: Here's an ugly hack to figure out whether items are in-flight or not:

SearchCommand
    .Do(x => Interlocked.Increment(ref searchesInFlight))
    .Select(x => Observable.Start(() => PerformSearch(x), RxApp.TaskPoolScheduler).Do(x => Interlocked.Decrement(ref searchesInFlight)))
    .Switch()        
    .ToProperty(this, x => x.SearchResults);