Problem combining Reactive Framework (Rx) queries to provide correct UI behaviour

Question

I'm trying to remove the more tradition event handlers from a Silverlight application in favour of using a number of Rx queries to provide a better, easier to manage, behavioural abstraction.

The problem that I need to solve, but can't quite crack it the way I want, is getting the behaviour of a search screen working. It's pretty standard stuff. This is how it should behave:

• I have a text box where the user can enter the search text.
• If there is no text (or whitespace only) then the search button is disabled.
• When there is non-whitespace text then the search button is enabled.
• When the user clicks search the text box and the search button are both disabled.
• When the results come back the text box and the search button are both enabled.

I have these observables (created via some extension methods over the standard events) to work with:

IObservable<IEvent<TextChangedEventArgs>> textBox.TextChangedObservable()
IObservable<IEvent<RoutedEventArgs>> button.ClickObservable()
IObservable<IEvent<LoadingDataEventArgs>> dataSource.LoadingDataObservable()
IObservable<IEvent<LoadedDataEventArgs>> dataSource.LoadedDataObservable()

I have these queries that work at the moment:

IObservable<bool> dataSourceIsBusy =
    dataSource.LoadingDataObservable().Select(x => true)
    .Merge(dataSource.LoadedDataObservable().Select(x => false));

IObservable<string> textBoxText =
    from x in textBox.TextChangedObservable()
    select textBox.Text.Trim();

IObservable<bool> textBoxTextIsValid =
    from text in textBoxText
    let isValid = !String.IsNullOrEmpty(text)
    select isValid;

IObservable<string> searchTextReady =
    from x in button.ClickObservable()
    select textBox.Text.Trim();

And then these subscriptions to wire it all up:

buttonIsEnabled.Subscribe(x => button.IsEnabled = x);
dataSourceIsBusy.Subscribe(x => textBox.IsEnabled = !x);
searchTextReady.Subscribe(x => this.ExecuteSearch(x));

(I do keep references to the disposables returned by the Subscribe methods. I have .ObserveOnDispatcher() added to each observable to ensure the code runs on the UI thread.)

Now while this works there is one thing that bugs me. The select statement in searchTextReady calls textBox.Text.Trim() to obtain the current trimmed search text, but I already have done this in the textBoxText observable. I really don't want to repeat myself so I would like to create a query that combines these observables and this is where I'm failing.

When I try the following query I get re-entrant calls to execute the search:

IObservable<string> searchTextReady =
    from text in textBoxText
    from x in button.ClickObservable()
    select text;

The following query it seems to work for the first query, but then each time I change the text in the text box thereafter the search is automatically executed without clicking the search button:

IObservable<string> searchTextReady =
    from text in button.ClickObservable()
        .CombineLatest(textBoxText, (c, t) => t)
    select text;

The following query requires a further text change to occur after the search button is clicked and then fails to run again:

IObservable<string> searchTextReady =
    from text in textBoxText
      .SkipUntil(button.ClickObservable())
      .TakeUntil(dataSource.LoadingDataObservable())
    select text;

Any ideas how I can make this work?

Answer 1

Have you tried ForkJoin

Something like:

    IObservable<string> searchTextReady = Observable.ForkJoin(textBoxText, button.ClickObservable());
    searchTextReady.Subscribe( ....

Answer 2

These kind of things are tricky on their own, so I ended up writing an M-V-VM + Rx library to help me out - it turns out that with this library, this task is pretty easy; here's the whole code, my blog explains more about how these classes work:

public class TextSearchViewModel
{
    public TextSearchViewModel
    {
        // If there is no text (or whitespace only) then the search button is disabled.
        var isSearchEnabled = this.ObservableForProperty(x => x.SearchText)
            .Select(x => !String.IsNullOrWhitespace(x.Value));

        // Create an ICommand that represents the Search button
        // Setting 1 at a time will make sure the Search button disables while search is running
        DoSearch = new ReactiveAsyncCommand(isSearchEnabled, 1/*at a time*/);

        // When the user clicks search the text box and the search button are both disabled.
        var textBoxEnabled = DoSearch.ItemsInflight
            .Select(x => x == 0);

        // Always update the "TextboxEnabled" property with the latest textBoxEnabled IObservable
        _TextboxEnabled = this.ObservableToProperty(textBoxEnabled, 
            x => x.TextboxEnabled, true);

        // Register our search function to run in a background thread - for each click of the Search
        // button, the searchResults IObservable will produce a new OnNext item
        IObservable<IEnumerable<MyResult>> searchResults = DoSearch.RegisterAsyncFunction(textboxText => {
            var client = new MySearchClient();
            return client.DoSearch((string)textboxText);
        });

        // Always update the SearchResults property with the latest item from the searchResults observable
        _SearchResults = this.ObservableToProperty(searchResults, x => x.SearchResults);
    }

    // Create a standard INotifyPropertyChanged property
    string _SearchText;
    public string SearchText {
        get { return _SearchText; }
        set { this.RaiseAndSetIfChanged(x => x.SearchText, value); }
    }

    // This is a property who will be updated with the results of an observable
    ObservableAsPropertyHelper<bool> _TextboxEnabled;
    public bool TextboxEnabled {
        get { return _TextboxEnabled.Value; }
    }

    // This is an ICommand built to do tasks in the background
    public ReactiveAsyncCommand DoSearch { get; protected set; }

    // This is a property who will be updated with the results of an observable
    ObservableAsPropertyHelper<IEnumerable<MyResult>> _SearchResults;
    public IEnumerable<MyResult> SearchResults {
        get { return _SearchResults.Value; }
    }
} 

Answer 3

DistinctUntilChanged is what you are looking for.

E.g. something like this should work:

// low level observables
var dataSourceLoading = ... // "loading" observable
var dataSourceLoaded = ... // "loaded" observable
var textChange = Observable.FromEvent<TextChangedEventArgs>(MyTextBox, "TextChanged");
var click = Observable.FromEvent<RoutedEventArgs>(MyButton, "Click");

// higher level observables
var text = textChange.Select(_ => MyTextBox.Text.Trim());
var emptyText = text.Select(String.IsNullOrWhiteSpace);
var searchInProgress = dataSourceLoading.Select(_ => true).Merge(dataSourceLoaded.Select(_ => false));

// enable/disable controls
searchInProgress.Merge(emptyText)
    .ObserveOnDispatcher()
    .Subscribe(v => MyButton.IsEnabled = !v);
searchInProgress
    .ObserveOnDispatcher()
    .Subscribe(v => MyTextBox.IsEnabled = !v);

// load data 
click
    .CombineLatest(text, (c,t) => new {c,t})
    .DistinctUntilChanged(ct => ct.c)
    .Subscribe(ct => LoadData(ct.t));