After Rx Road show

Last week I have finished Rx Road show. I have been giving presentations about Reactive Framework on four Polish .Net Users Groups: in Wroclaw, Lodz, Krakow and Chorzow. Thank you all for coming to my session and filling the evaluations, I hope you have learned something useful about Rx during the session. I have also good news for those of you who were interested in my presentation but for some reason couldn’t come, one of the presentation has been recorded and it is now available for download or watching live online here. Slides and all code samples can be download from here. Enjoy!

RxSandbox V1

[New version (2011.05.06)]

[RxSandbox downlaod has been upgraded to the newest version of Rx (Build 1.0.2677.0 08/27/2010).] download

[RxSandbox downlaod has been upgraded to the newest version of Rx (Build 1.0.2617.0 07/15/2010).] download
(Changes: solution converted to VS2010, Net 4.0; 56 operators, grouping operators on the tree control; zoom)

[RxSandbox downlaod has been upgraded to the newest version of Rx (Build 1.0.2441.0 04/14/2010).] download

[RxSandbox downlaod has been upgraded to the newest version of Rx (Build 1.0.2350.0 03/15/2010).] download

I am really happy to announce that the new version of RxSandbox has been just released. Last time I have been writing about the concept and the main ideas behind the project. That was just a small prototype, the proof of concept. This release is much more mature, the user can test more standard Rx operators and the API has been changed a little bit, however the concept stated the same.

These are the main new features:

• Marble diagrams
• New powerful API
• Extensibility mechanism
• Description of Rx the operators taken from Rx documentation

Lets start from the the end-user which is not necessary interested in writing any code. He just wants to experiment with Rx operators, check how they behave in specific scenarios. When we start RxSandbox application we will see the list of standard operators on the left hand side. Lets assume that we don’t know how Merge operator works. After double click on the Merge node the new tab will be displayed. We can find the short description of Merge operator taken from documentation provided by Rx installer, we can also find the code sample that can be tested interactively trough UI. Each input argument is presented as a simple automatically generated UI with one textbox where we can write input value, tree buttons and the history of the source. The output of the expression can be presented in two ways: by a simple list displaying the results (“Output” tab) or the marble diagrams drawn live during testing (‘Marble diagram – Live” tab). There is also one tab called “Marble diagram - Definition” showing the operator definition.

Now lets see what the developer can do with RxSandbox. The most important class in Rx API is the ExpressionDefinition. It holds all necessary information describing tested expression such as name, description and sample code.

When we look inside RxSandbox project we will see that the definitions of all standard operators are very similar, for example the Merge operator is defined like this:

public static ExpressionDefinition Merge()
{
    Expression<Func<IObservable<string>, IObservable<string>, IObservable<string>,
        IObservable<string>>> expression
            = (a, b, c) => Observable.Merge(a, b, c);
    
    return ExpressionDefinition.Create(expression);
}

This is all we need to write. Other things like operator’s name, description and the text of expression can be inferred from the Linq expression. Of course all these information can be set manually using appropriate Create method overload and ExpressionSettings class. All .Net types are supported as observable type, not only the System.String type like in this example. The only requirement is there must exist a TypeConverter for that type. Later in this post I’ll how implement TypeConverter for custom type and how to create ExpressionDefinition without using Linq expression but using the whole method with many statements. The second very important class in RxSandbox API is an ExpressionInstance class which is very useful in scenarios where we want to use some RxSandbox functionalities directly from code without any UI experience (for example during writing Unit Tests or recording marble diagram).

Expression<Func<IObservable<string>, IObservable<string>, IObservable<string>,
   IObservable<string>>> expression
       = (a, b, c) => Observable.Merge(a, b, c);

ExpressionDefinition definition = ExpressionDefinition.Create(expression);

using (var instance = ExpressionInstance.Create(definition))
{
    ExpressionInstance instance = ExpressionInstance.Create(definition);

    // using non-generic type 'ObservableSource'
    ObservableSource output1 = instance.Output;
    output1.ObservableStr.Subscribe(Console.WriteLine);

    // using generic type 'ObservableSource<T>'
    ObservableOutput<string> output2 = instance.Output as ObservableOutput<string>;
    output2.Observable.Subscribe(Console.WriteLine);

    instance["a"].OnNext("one");    // using non-generic type 'ObservableInput'
    (instance["a"] as ObservableInput<string>).OnNext("two"); // using generic type
    instance["b"].OnNext("tree");
    instance["a"].OnCompleted();
    instance["b"].OnCompleted();
    instance["c"].OnCompleted();
}

When we add delay before sending each input signal we can very easily record sample marble diagram.

internal static class Extensions
{
    internal static void OnNext2(this ObservableInput input, string value)
    {
        input.OnNext(value);
        Thread.Sleep(100);
    }
    internal static void OnError2(this ObservableInput input)
    {
        input.OnError(new Exception());
        Thread.Sleep(100);
    }
    internal static void OnCompleted2(this ObservableInput input)
    {
        input.OnCompleted();
        Thread.Sleep(100);
    }
}

Marble diagrams are described in a very simple object model.

This object model can be serialized to Xml format, for instance the code above creates fallowing marble diagram:

<Diagram>
  <Input Name="a">
    <Marble Value="one" Order="0" />
    <Marble Value="two" Order="1" />
    <Marble Kind="OnCompleted" Order="3" />
  </Input>
  <Input Name="b">
    <Marble Value="tree" Order="2" />
    <Marble Kind="OnCompleted" Order="4" />
  </Input>
  <Input Name="c">
    <Marble Kind="OnCompleted" Order="5" />
  </Input>
  <Output>
    <Marble Value="one" Order="0" />
    <Marble Value="two" Order="1" />
    <Marble Value="tree" Order="2" />
    <Marble Kind="OnCompleted" Order="5" />
  </Output>
</Diagram>

As we can see single marble diagram have a very simple Xml representation and diagrams for all standard operators from RxSandbox are stored in Diagrams.xml file (this file path can be changed in the configuration files).

Short description added to all standard operators extracted from Rx documentation Xml file is a next new feature of current release. Of course not all tested reactive expression are related to one particular operator so the description can be set manually (ExpressionSettings.Description property). When we want write a very complicated Linq expression or the expression is passed as delegate type to the ExpressionDefinition.Create method and we also want to provide description from particular Rx operator at the same time, we can do it indicating operator thought MethodInfo type (ExpressionSettings.Operator property).

The last but not least new feature is the extensibility mechanism. When we want to write our custom reactive expression without changing anything inside RxSadbox project we can do it by implementing IExpressionProvider interface directly or by inheriting from an abstract class ExpressionAttributeBasedProvider and setting our assembly name in the configuration file (ExtensionsAssembly element). During startup process RxSandbox loads that assembly, analyzes it and finds all expression providers. Few weeks ago I have been writing about Incremental Find implemented using Rx, lets see how such a query can tested via RxSandbox.

[AttributeUsage(AttributeTargets.Method,AllowMultiple = false, Inherited = true)]
public class ExpressionAttribute : Attribute { }

public interface IExpressionProvider
{
    IEnumerable<ExpressionDefinition> GetExpressions();
}

public abstract class ExpressionAttributeBasedProvider : IExpressionProvider
{
    public IEnumerable<ExpressionDefinition> GetExpressions()
    {
        var q =
            from m in this.GetType().GetMethods()
            let attr = Attribute.GetCustomAttribute(m, typeof(ExpressionAttribute)) 
                as ExpressionAttribute
            where attr != null
            select m.Invoke(null, null) as ExpressionDefinition;
        return q.ToList();
    }
}

public class CustomExpressions : ExpressionAttributeBasedProvider
{
    [Expression]
    public static ExpressionDefinition IncrementalSearch()
    {
        Func<IObservable<string>, IObservable<Person>, IObservable<Person>> expression
                = (codeChanged, webServiceCall) =>
                      {
                        var q =
                            from code in codeChanged
                            from x in Observable.Return(new Unit())
                                .Delay(TimeSpan.FromSeconds(4)).TakeUntil(codeChanged)
                            from result in webServiceCall.TakeUntil(codeChanged)
                            select result;

                          return q;
                      };

        return ExpressionDefinition.Create(expression, new ExpressionSettings
           {
               Name = "Incremental find",
               Description = @"Send the code of the person you are looking for, "
                    + "after four seconds (if you don't send new code again) web service "
                    + "will be called. The result won't be returned if new code is provided "
                    + "in the meantime.",                   
           });
    }
}

[TypeConverter(typeof(PersonConverter))]
public class Person
{
    public string Code { get; set; }
    public string Name { get; set; }
}

public class PersonConverter : TypeConverter
{
    public override bool CanConvertFrom(ITypeDescriptorContext context, Type sourceType)
    {
        if (sourceType == typeof (string))
            return true;
        return base.CanConvertFrom(context, sourceType);
    }
    
    public override object ConvertFrom(ITypeDescriptorContext context, CultureInfo culture, object value)
    {
        if (value is string)
        {
            string[] v = ((string)value).Split(new[] { ',' });
            return new Person {Code = v[0], Name = v[1]};
        }
        return base.ConvertFrom(context, culture, value);
    }
    
    public override object ConvertTo(ITypeDescriptorContext context,
       CultureInfo culture, object value, Type destinationType)
    {
        if (destinationType == typeof (string))
        {
            var person = value as Person;
            return person.Code + "," + person.Name;
        }                
        return base.ConvertTo(context, culture, value, destinationType);
    }
}

So that’s it for this release of RxSandbox. I encourage you to play with it a little bit and let me know what you think.

And that’s not the end of the project. In upcoming releases I plan to:

• create Silverlight version with ability to write reactive expression directly in the web browser
• add integration with MEF
• add better look and feel experience

Here you can download sources and binaries

After Microsoft Technology Summit 2009

I would like to thank all of the attendees who came to my session - "Workflow Foundation 4.0" on the MTS2009 conference. Thanks for all evaluations and comments! The videos from the conference has just been published. Here is my webcast presentation (polish version).

RxSandbox

Rx is awesome :)

I have just started a new project called RxSandbox. It will be a very simple application that help to understand how Rx works, how each operator is implemented. But lets start from the beginning. Few days ago I was playing with Rx operators such as Zip, Merge, Wait, etc. writing a very short chunk of code for each operator. I was wondering how do they work in details, if they cache the values from sources, when exactly is OnCompleted executed on the observers and so on. The code could look like this:

a.Zip(b, (x, y) => x + " - " + y)

I wanted to a and b be the implementation of IObservable<string>. But what is the easiest way to create the implementation for them? I just wanted to be focused on writing code using operators so I needed some infrastructure for creating sample observable data sources. The RxSandbox is the solution! When you write something like this:

Expression<Func<ManualObservable<string>, ManualObservable<string>,
    IObservable<string>>> zipExpression
        = (a, b) => a.Zip(b, (x, y) => x + " - " + y);
        
Control control = RxExpressionVisualizer.CreateControl(zipExpression);

RxSandbox will automatically generate testing UI control:

In the code above ManualObservable means that the user can manually define the values sent to the expression by writing them in a text box. Other possible could be RandomObservable, IntervalObservable and there are many more options. The UI control will be different for each type of Observable source. The only reason why the Expression<Func<...>> type has been used here is that this allows us to display the body of expression, but of course this is just an option. In general, the test case is a method taking some Observable sources and returning observable collection. It can be some really complicated code snippet with many statements as well.

Possible scenarios working with RxSanbox:

• Run RxSandbox application, create new project in VS, write Rx expression, compile it, RxSandbox automatically  finds new version of dll file and loads it, you are ready to test it (it may be MEF, VS AddIn not just of stand along application)
• Run RxSandbox, write your Rx expression inside RxSandbox, start testing it (the easy way to do this is to use ExpressionTextBox control from WF4.0 which allows us to write single VB expression and expose it as an Linq expression type)
• Drawing marble diagrams live during testing and also presenting the definitions of the operators as marble diagrams (of course with pac-mans and hearts :) - )
• Supporting Silverlight version
• Many many more....

I hope you got the idea. Here you can find the first prototype.

Puzzle

What will be displayed on the screen when we write it (and why) ?

Action<string> a = Console.WriteLine;
var b = a;
var c = (a += Console.WriteLine);

a("Rx");
b("rulez");
c("!");

Hint: delegate type is:
- reference type
- immutable type

Linq to ICollectionView

Currently I'm working on the project written in Silverlight and I have encountered an interface called ICollectionView and its standard implementation PagedCollectionView. In short, this components allow us to create the view of collection of items with filtering, grouping and sorting functionality. The idea behind this interface is very similar to DataView/DataTable mechanism. DataTable is responsible for storing data and DataView is just an appropriately configured proxy (only filtering and sorting in this case) which can be bound to UI controls. Let's look at a very simple example:

public class Number
{
    public int Value { get; set; }
    public int Random { get; set; }

    public static Number[] GetAll()
    {
        var random = new Random();
        return
            (from n in Enumerable.Range(1,5)
             from m in Enumerable.Repeat(n, n)
             select new Number {Value = n, Random = random.Next(10)}).ToArray();            
    }
}

// Silverlight
public class LinqToICollectionView : UserControl
{
    public LinqToICollectionView()
    {
        Number[] numbers = Number.GetAll();

        Content = new StackPanel
        {
            Orientation = Orientation.Horizontal,
            Children =
            {
                new DataGrid { ItemsSource = new PagedCollectionView(numbers).SetConfiguration()},
            }
        };
    }
}

public static class Configurator
{
    public static ICollectionView SetConfiguration(this ICollectionView view)
    {
        // filtering
        view.Filter = (object o) => ((Number)o).Value < 5;
        // grouping
        view.GroupDescriptions.Add(new PropertyGroupDescription("Value"));
        // sorting
        view.SortDescriptions.Add(new SortDescription("Value", ListSortDirection.Descending));
        view.SortDescriptions.Add(new SortDescription("Random", ListSortDirection.Ascending));
        return view;
    }     
}

But wait a minute, we said ... filtering, ordering, grouping ? Let's use LINQ query to configure ICollectionView:

public static class Configurator
{
    public static ICollectionView SetConfigurationWithLinq(this ICollectionView view)
    {
        var q =
            from n in new View<Number>()
            where n.Value < 5
            orderby n.Value descending, n.Random
            group n by n.Value;
        q.Apply(view);
        return view;
    }        
}

The whole implementation consists of 3 simple classes: View<T>, OrderedView<T> and GroupedView<T>.

public class View<T>
{
    public IEnumerable<GroupDescription> GroupDescriptions { get { ... } }
    public IEnumerable<SortDescription> SortDescriptions { get { ... } }
    public Func<T,bool> Filter { get { ... } }
    
    public View<T> Where(Func<T, bool> func) { ... }
    public SortedView<T> OrderBy<T2>(Expression<Func<T, T2>> func) { ... }
    public SortedView<T> OrderByDescending<T2>(Expression<Func<T, T2>> func) { ... }
    public GroupedView<T> GroupBy<T2>(Expression<Func<T, T2>> func) { ... }
    
    public void Apply(ICollectionView collectionView) { ... }
}
public sealed class SortedView<T> : View<T>
{    
    public SortedView<T> ThenBy<T2>(Expression<Func<T, T2>> func) { ... }
    public SortedView<T> ThenByDescending<T2>(Expression<Func<T, T2>> func)  { ... }
}
public sealed class GroupedView<T> : View<T>
{
    public GroupedView<T> ThenBy<T2>(Expression<Func<T, T2>> func) { ... }
}

Methods for sorting and grouping which take an expression tree as a parameter analyze the tree looking for indicated members (fields or properties) and collect appropriate SortDescription and GroupDescription objects. Where method takes a delegate type which is combined via logical and operator with existing filter delegate set previously (in case when Where method is called many times). Of course the same mechanism work also in WPF.

Number[] numbers = Number.GetAll();

// WPF
new Window
{
    Content = new StackPanel
    {
        Orientation = Orientation.Horizontal,
        Children =
        {
            CreateListView(new CollectionViewSource { Source = numbers }.View.SetConfiguration()),
            CreateListView(new CollectionViewSource { Source = numbers }.View.SetConfigurationWithLinq())
        }
    }
}
.ShowDialog();


private static ListView CreateListView(ICollectionView view)
{
    return new ListView
    {
        GroupStyle = { GroupStyle.Default },
        View = new GridView
        {
            Columns =
            {
                new GridViewColumn { Header = "Value", DisplayMemberBinding = new Binding("Value") },
                new GridViewColumn { Header = "Random", DisplayMemberBinding = new Binding("Random") },
            }
        },
        ItemsSource = view
    };
}

At the end I'd like to mention one interesting thing. We only support filtering, sorting and grouping and don't support for instance projection, joining and so on. That's way this code should compile:

var v1 = // only filtering specified
    from n in new View<Number>() 
    where n.Value < 5 
    select n;

var v2 = // grouping (last grouping definition overrides previous ones)
    from n in new View<Number>()
    group n by n.Random into s 
    group s by s.Value;

but this will not:

var v3 = // joining is not supported
    from p in new View<Number>()
    join pp in new[] { 1, 2, 3, 4 } on p.Random equals pp
    select p;

var v4 = // projection is not supported
    from p in new View<Number>()
    where p.Value > 5
    select p.Random;

var v5 = // at least one filtering, grouping or sorting definition must be specified
    from p in new View<Number>()
    select p;

var v6 = // Numer type is the only valid type of grouped element
    from p in new View<Number>()
    group p.Random by p.Value;

As a homework I leave you a question: why it works this way ? :)

Sources

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 =
@"
http://channel9.msdn.com/shows/Going+Deep/Expert-to-Expert-Brian-Beckman-and-Erik-Meijer-Inside-the-NET-Reactive-Framework-Rx/

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())).
            ToArray();

        subject.OnNext(result);
        subject.OnCompleted();
    }, 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}",
        r.text,
        r.results.Length,
        string.Join("\n", r.results));
});


form.ShowDialog();

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

Sources