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

I'm trying to remove the mor开发者_运维问答e 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?

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; }
    }
} 

Have you tried ForkJoin

Something like:

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

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));

Thanks for everyone's help on this question. I couldn't give anyone the answer because I found that the solution was to use the .Switch() operator and no-one gave that answer.

Since I did find the answer to my problem I thought I would post it here to help anyone else trying to do what I've done.

Part of my original problem what that I didn't want to repeat code. So I introduced a Func<string, bool> textIsValid that could be re-used in my IObservable<bool> textBoxTextIsValid and in my desired IObservable<string> searchTextReady.

Func<string, bool> textIsValid = t => !String.IsNullOrEmpty(t);

Now searchTextReady can be defined as follows:

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

So .Switch() works by taking an IObservable<IObservable<string>> and flattening it out into an IObservable<string>.

It's easy to create an IObservable<IObservable<string>> - just write a query against an observable and select some other observable.

For example:

IObservable<Unit> xs= ...;
IObservable<string> ys= ...;

IObservable<IObservable<string>> zss = xs.Select(x => ys);

IObservable<string> zs = zss.Switch();

But this doesn't do anything particularly useful.

If it's changed around as:

IObservable<Unit> clicks = ...;
IObservable<string> texts = ...;
Func<string, bool> isValid = ...;

IObservable<IObservable<string>> zss =
    texts.Select(t =>
        clicks
            .Take(1)
            .TakeUntil(texts)
            .Where(x => isValid(t))
            .Select(x => t));

IObservable<string> zs = zss.Switch();

The final observable can be described as:

"Whenever the text changes take only the next click provided that the text doesn't change again and such that the text is valid and then select the value that the text changed to."

I hope this answer helps someone else.