How can I combine IObservable<T>.Throttle() with some other event source using Reactive Extensions?

Question

I have a simple search field which should search automatically after the user has stopped typing or when he presses Search. The first part can be easily achieved with the following

var inputs = Observable.FromEventPattern<SomeEventArgs>(searchBar, "TextChanged")
                       .Select(pattern => pattern.EventArgs.SearchText)
                       .Throttle(TimeSpan.FromMilliseconds(500));

I can also chain that to do the actual search like this

var results = from query in inputs
              from results in Observable.FromAsync<Something>(() => Search(query))
              select results;

But the problem is that there is no way to skip ahead when the user presses the search button. From my understanding of Rx the code should probably be something like this:

// inputs = the same event stream without Throttle
// buttonClicks = construct event stream for search button clicks

// ??? somehow make a third stream which lets a value through
// ??? only after a delay and when the value hasn't changed,
// ??? OR when the second stream yields a value

// async search

I can see how I would write this imperatively by using something like Stopwatch and resetting it as the user types, and if a click comes through I could just skip it over. But in the world of Rx it would probably look like (pardon the pseudo-linq-code)

from query in inputs
where (query.isLast() and query.timestamp > 500.ms.ago) or buttonClicked
...

I need to be able to get through the last query input immediately if the second event source yields a value, or if there is no value then just wait a specified delay as if Throttle is used.

Answer 1

First off, the typical search Rx looks like this:

var searchResults = Observable.FromEventPattern<SomeEventArgs>(searchBar, "TextChanged")
                              .Select(pattern => pattern.EventArgs.SearchText)
                              .Throttle(TimeSpan.FromMilliseconds(500))
                              .DistinctUntilChanged()
                              .Select(text => Observable.Start(() => Search(text)))
                              .Switch()

Select gives you a stream of result streams, Switch will return the most recently created stream. I added DistinctUntilChanged to prevent duplicate queries being submitted.

One strategy for what you have described is to supply the throttle a throttle duration selector that will emit either after the throttle duration or if a button is clicked. I've knocked together a sample ViewModel that avoids using any libraries other than Rx 2.1 to show how this might be done. This is the whole ViewModel - I'll leave the View and Repository to your imagination, but it should be clear what they do.

Final caveat - I've tried to keep this sample brief and leave out unnecessary details that might cloud understanding so this isn't production ready:

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Runtime.CompilerServices;
using System.Windows.Input;
using StackOverflow.Rx.Annotations;
using StackOverflow.Rx.Model;

namespace StackOverflow.Rx.ProductSearch
{
    public class ClassicProductSearchViewModel : INotifyPropertyChanged
    {
        private string _query;
        private IProductRepository _productRepository;
        private IList<Product> _productSearchResults;

        public ClassicProductSearchViewModel(IProductRepository productRepository)
        {
            _productRepository = productRepository;
            // Wire up a Button from the view to this command with a binding like
            // <Button Content="Search" Command="{Binding ImmediateSearch}"/>
            ImmediateSearch = new ReactiveCommand();

            // Wire up the Query text from the view with
            // a binding like <TextBox MinWidth="100" Text="{Binding Query, UpdateSourceTrigger=PropertyChanged}"/>
            var newQueryText = Observable.FromEventPattern<PropertyChangedEventHandler, PropertyChangedEventArgs>(
                h => PropertyChanged += h,
                h => PropertyChanged -= h)
                .Where(@event => @event.EventArgs.PropertyName == "Query")
                .Select(_ => Query);

            // This duration selector will emit EITHER after the delay OR when the command executes
            var throttleDurationSelector = Observable.Return(Unit.Default)
                                                     .Delay(TimeSpan.FromSeconds(2))
                                                     .Merge(ImmediateSearch.Select(x => Unit.Default));


            newQueryText
                .Throttle(x => throttleDurationSelector)
                .DistinctUntilChanged()
                /* Your search query here */
                .Select(
                    text =>
                        Observable.StartAsync(
                            () => _productRepository.FindProducts(new ProductNameStartsWithSpecification(text))))
                .Switch()
                .ObserveOnDispatcher()
                .Subscribe(products => ProductSearchResults = new List<Product>(products));
        }

        public IList<Product> ProductSearchResults
        {
            get { return _productSearchResults; }
            set
            {
                if (Equals(value, _productSearchResults)) return;
                _productSearchResults = value;
                OnPropertyChanged();
            }
        }

        public ReactiveCommand ImmediateSearch { get; set; }

        public string Query
        {
            get { return _query; }
            set
            {
                if (value == _query) return;
                _query = value;
                OnPropertyChanged();
            }
        }

        public event PropertyChangedEventHandler PropertyChanged;

        [NotifyPropertyChangedInvocator]
        protected virtual void OnPropertyChanged([CallerMemberName] string propertyName = null)
        {
            PropertyChangedEventHandler handler = PropertyChanged;
            if (handler != null) handler(this, new PropertyChangedEventArgs(propertyName));
        }
    }

    // A command that is also an IObservable!
    public class ReactiveCommand : ICommand, IObservable<object>
    {
        private bool _canExecute = true;
        private readonly Subject<object> _execute = new Subject<object>();

        public ReactiveCommand(IObservable<bool> canExecute = null)
        {
            if (canExecute != null)
            {
                canExecute.Subscribe(x => _canExecute = x);
            }
        }

        public bool CanExecute(object parameter)
        {
            return _canExecute;
        }

        public void Execute(object parameter)
        {
            _execute.OnNext(parameter);
        }

        public event EventHandler CanExecuteChanged;

        public IDisposable Subscribe(IObserver<object> observer)
        {
            return _execute.Subscribe(observer);
        }
    }
}

There are libraries out there like Rxx and ReactiveUI that can make this code simpler - I've not used them here so there is minimal "magic" going on!

My ReactiveCommand in this example is a simple implementation of one included in ReactiveUI. This looks like a command and an IObservable at the same time. Whenever it is executed it will stream the command parameter.

Here is an example using ReactiveUI from the author's blog: http://blog.paulbetts.org/index.php/2010/06/22/reactivexaml-series-reactivecommand/

In another answer I look at the variable throttling feature in isolation.