Sunday, August 30, 2009

Incremental find with Reactive Framework

This time we will implement "incremental find" using Reactive Framework. This is a very good example showing what the Rx is all about.

Lets say we have a Windows Forms application with a single TextBox. When the user stops typing, the application immediately sends the request to the remote web service to find all words that containt the text from the TextBox. We use Rx to handle two important issues:

  • how to find the moment when user has just stopped typing ?
  • how to ensure that the correct results will be displayed when calling web service is implemented as asynchronous operation (results for the most recent input should discard responses for all previous requests) ?

For the sake of simplicity we simulate calling a web service by a simple asynchronous operation returning results after 3 or 6 seconds. This allows us to easily check whether the correct results are displayed every time. Just type 'iobserv' then wait for about 2 seconds (user stopped writing) and append letter 'e'. After about 3 second the results for 'ibserve' text should be displayed and then never changed.

const string RxOverview =

Now, what is Rx?

The .NET Reactive Framework (Rx) is the mathematical dual of LINQ to Objects. It consists of a pair 
of interfaces IObserver / IObservable that represent push-based, or observable, collections, plus a 
library of extension methods that implement the LINQ Standard Query Operators and other useful 
stream transformation functions.
Observable collections capture the essence of the well-known subject/observer design pattern, and 
are tremendously useful for dealing with event-based and asynchronous programming, i.e. AJAX-style 
applications. For example, here is the prototypical Dictionary Suggest written using LINQ query 
comprehensions over observable collections:

IObservable<Html> q = from fragment in textBox
               from definitions in Dictionary.Lookup(fragment, 10).Until(textBox)
               select definitions.FormatAsHtml();

q.Subscribe(suggestions => { div.InnerHtml = suggestions; })

var textBox = new TextBox();
var label = new Label
                    Text = "results...",
                    Location = new Point(0, 40),
                    Size = new Size(300, 500),
                    BorderStyle = BorderStyle.FixedSingle,                                
var form = new Form { Controls = { textBox, label } };

Func<string, IObservable<string[]>> search = (s) =>
    var subject = new Subject<string[]>();

    ThreadPool.QueueUserWorkItem((w) =>
        Thread.Sleep(s.Length % 2 == 0 ? 3000 : 6000);

        var result = RxOverview.
            Split(new[] { " ", "\n", "\t", "\r" }, StringSplitOptions.RemoveEmptyEntries).
            Where(t => t.ToLower().Contains(s.ToLower())).

    }, null);

    return subject;

IObservable<Event<EventArgs>> textChanged = textBox.GetObservableTextChanged();

var q =
    from e in textChanged
    let text = (e.Sender as TextBox).Text
    from x in Observable.Return(new Unit()).Delay(1000).Until(textChanged)  // first issue
    from results in search(text).Until(textChanged)                         // second issue
    select new { text, results };

var a1 = q.Send(SynchronizationContext.Current).Subscribe(r =>
    label.Text = string.Format(" Text: {0}\n Found: {1}\n Results:\n{2}",
        string.Join("\n", r.results));


Now try to implement the same functionality without the Rx Framework.



Jeffrey said...

Nice post!

The search function can be simplified like this:

var search = Observable.ToAsync < string,string[]>(s=>
var result = RxOverview.
Split(new[] { " ", "\n", "\t", "\r" }, StringSplitOptions.RemoveEmptyEntries).
Where(t => t.ToLower().Contains(s.ToLower())).

return result;



Marcin Najder said...

Hi Jeffrey,

Thanks for your comment .

You are absolutely right, in Windows Forms this solutions works fine. The idea for the post comes from a project I'm currently working on, where I wanted to use Incremental Find. The project is written in Silverlight and when you try to port your solution to Silverlight You will encounter a problem. Calling delegate function created by "Observable.ToAsync<>()" throws NotSupportedException. The StackTrace shows that the "Delegate.BeginInvoke" method is invoked underneath but this method is not supported in Silverlight.

var m = Observable.ToAsync(() => "");
m(); // throws exception
//((Func< string >) (() => "")).BeginInvoke(null, null); // throws exception
catch (NotSupportedException exception)
Debug.WriteLine(exception.Message); // "Specified method is not supported."

Even if I leave my solution though, the whole LINQ query dosen't work properly either. I'm still getting some strange exception. So far mashalling query execution to the UI thread is the only solution I have found.

var q =
from e in textChanged
let text = (e.Sender as TextBox).Text
from x in Observable.Return(new Unit()).Delay(1000).Post(uiContext).Until(textChanged)
from results in search(text).Post(uiContext).Until(textChanged)
select new { text, results };

Jeffrey van Gogh said...

Wow, nice bug! Didn't realize Delegate.BeginInvoke was not supported in Silverlight. We'll fix this asap..

I'll also look into the second issue you're seeing.



dinesh said...

Observable.Start(() =>
return "Hello There";
}).Subscribe(Value => button.Content = Value);

I am working on silverLight application.

above code raise this error
please send me solution of this error


Error :Specified method is not supported.

Stack Trace :
at System.Func`1.BeginInvoke(AsyncCallback callback, Object object)
at System.Func`3.Invoke(T1 arg1, T2 arg2)
at System.Linq.Observable.<>c__DisplayClass10f`1.ToAsync>b__10d()
at System.Linq.Observable.Start[T](Func`1 function)
at SilverlightApplication11.MainPage.<.ctor>b__0(Object o, RoutedEventArgs e)
at System.Windows.Controls.Primitives.ButtonBase.OnClick()
at System.Windows.Controls.Button.OnClick()
at System.Windows.Controls.Primitives.ButtonBase.OnMouseLeftButtonUp(MouseButtonEventArgs e)
at System.Windows.Controls.Control.OnMouseLeftButtonUp(Control ctrl, EventArgs e)
at MS.Internal.JoltHelper.FireEvent(IntPtr unmanagedObj, IntPtr unmanagedObjArgs, Int32 argsTypeIndex, String eventName)

Marcin Najder said...


I have just run your code and it works fine. Check if you are using the latest version of Rx ( and SL3.