LeeCampbell

The definitive list of Rx sites

2011-06-21T16:22:00.001+01:00

Over at the Rx forums, Eamon_OTuathail has created what seems to be the list of Rx sites. It is a list of Blogs and open source projects.

Great stuff. This should save a lot of time for a lot of people.

http://social.msdn.microsoft.com/Forums/en-US/rx/thread/2cbd3b1c-d535-46ba-a9cf-3cd576a8e7c2

Humbled to make the grade.

Isolating custom Library dependencies versions from consumer dependency versions

2011-06-18T12:00:00.001+01:00

This post is more about the CLR and dependency management than it is about Rx. However at this point in Rx’s lifecycle it seems relevant to comment on. The same principles could obviously be applied to open source projects et. al.

I have recently heard of an interesting situation that no-doubt has proved troublesome to people before. This problem is particularly interesting with Rx and it’s recent rate of versions. If you are trying to incorporate Rx into a library of yours and you then want to publish that library, you are effectively forcing users to use the same version of Rx that you do. Considering that a version of Rx comes out say every two months and that often there are breaking changes, this can create quite a mess. It is also made more interesting that up until recently they have not specified if a release was experimental or considered stable.

To provide an example to better understand my particular point; imagine you have a library that wraps a messaging platform. You want to avoid the use of Events and APM, and expose things via IObservable<T>. You feel happy that IObservable<T> is exposed natively in .NET 4 so you should not have to expose your implementations of Rx. You do however, want to use Rx as it has features you need and don’t want to (re-)write yourself. Your standard approaches to package/deployment are:

deploy the parts of the Rx libraries you use with your code. Users can just put them all into a “lib\MyFramework” folder and reference them.
excluded the libraries from your code and set Specific Version = False and hope your code will work with the consumer’s version of Rx
use a package management tool like Nuget to publish your package and specify the valid versions of Rx your library will work with.
rely on things being in the GAC so you can utilise the side-by-side versioning it provides
just try and implement the parts of Rx you want and avoid DLL Hell.

I gave this some thought and I think I have come up with a solution that could help library authors protect themselves. While there is the obvious option of using Nuget as part of your dependency management, this does not solve the problem, it just eases the pain. If the customer wants to use the latest version of Rx and you only support the 3 previous versions, your customer is still in some trouble.

The theory i had was that I can specify the specific version of Rx I want to reference in my library, the problem being that it may be named the same as the client’s referenced version. Depending on where their references were built to, they could overwrite each other. It seemed the solution was to embed the dependency into my library.

This turns out to actually be quite easy. If you have a project in visual studio that references your version of Rx, you have to follow these steps:

Ensure you have a file reference (not a project or a GAC reference) to the dependency, in this case System.Reactive.dll
Set the reference to be Specific Version = True
Set the reference Copy Local = False
Embed the dependency into your library. I created a folder in my project called EmbeddedAssemblies. I “Add Existing…” to this folder, navigate to the dependencies (just System.Reactive.dll in this case), and then choose “Add as Link…”
Set the “Build Action” of the newly added link to Embedded Resource
Ensure that the resource is loaded correctly at run time…

The last part of that list proves to be not too hard. You can hook on to the AppDomain.AssemblyResolve event and load your embedded resource. You can return your embedded dependency by reading the byte stream and creating the assembly from it and then returning that in the event handler

internal static class DependencyResolver
{
  private static int _isSet = 0;

  internal static void Ensure()
  {
    if (Interlocked.CompareExchange(ref _isSet, 1, 0) == 0)
    {
      var thisAssembly = Assembly.GetExecutingAssembly();
      var assemblyName = new AssemblyName(thisAssembly.FullName).Name;
      var embededAssemblyPrefix = assemblyName + ".EmbeddedAssemblies.";

      var myEmbeddedAssemblies =
        Assembly.GetExecutingAssembly().GetManifestResourceNames()
          .Where(name => name.StartsWith(embededAssemblyPrefix))
          .Select(resourceName =>
                    {
                      using (var stream = Assembly.GetExecutingAssembly().GetManifestResourceStream(resourceName))
                      {
                        var assemblyData = new Byte[stream.Length];
                        stream.Read(assemblyData, 0, assemblyData.Length);
                        return Assembly.Load(assemblyData);
                      }
                    })
          .ToDictionary(ass => ass.FullName);

      AppDomain.CurrentDomain.AssemblyResolve += (sender, args) =>
                                                    {
                                                      Assembly assemblyToLoad = null;
                                                      myEmbeddedAssemblies.TryGetValue(args.Name, out assemblyToLoad);
                                                      return assemblyToLoad;
                                                    };
    }
  }
}

You can return null in the event handler to say I don't know how to load this assembly. This allows others to be able to have a go at loading the assembly in the same way.

Next, to ensure that my embedded dependency resolver is called, I opted to make a call to it in the static constructor of the key types in my library. eg

public class SomeProvider
{
  static SomeProvider()
  {
    DependencyResolver.Ensure();
  }

  //other stuff goes here...
}

Have a look at the example code by downloading this zip file

Caveats : This is a thought and the code and concepts are only demo quality. I have not used this in production quality code. I also have not verified that the license allows for this style of packaging.

Further links:

Using different version of the same dll in one application – extern alias keyword
Best Practice for Assembly Loading – MSDN
Questions around the stability and license for Rx releases
Embed application’s depended dlls – Jeffery Richter

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Rx v.1.0.10425–Breaking changes!

2011-06-05T23:42:00.000+01:00

For those that follow Rx, many will have noticed a new drop was made available in late April 2011. This was interesting for numerous reasons:

I believe that it was the first drop since the Rx team move off of the dev labs site and on to the Microsoft Data site
They have supplied two downloads, Supported and Experimental
It is available via Nuget
It broke most of the sample code of this blog
...It has massive breaking changes !

The first thing I noticed was that only the more recent frameworks were in the “supported” release i.e. .NET 4, Silverlight 4 and Windows Phone. Next, when you open up the install directory (which has moved, again) the files are different to previous releases.

This is a comparison of the location and files of the old v.s. new version (.NET 4 target):

C:\Program Files\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2856.0\Net4
System.CoreEx.dll
System.Interactive.dll
System.Linq.Async.dll
System.Reactive.ClientProfile.dll
System.Reactive.dll
System.Reactive.ExtendedProfile.dll
System.Reactive.Testing.dll

C:\Program Files\Microsoft Reactive Extensions SDK\v1.0.10425\Binaries\.NETFramework\v4.0
Microsoft.Reactive.Testing.dll
System.Reactive.dll
System.Reactive.Providers.dll
System.Reactive.Windows.Forms.dll
System.Reactive.Windows.Threading.dll

Note the files names are named System.Reactive instead of hijacking the existing BCL namespaces such as System, System.IO, System.Linq etc. Here are a list of the old and new Namespaces found in the Rx libraries with the count of the Types in each namespace.

C:\Program Files\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2856.0\Net4
System                         6
System.Collections.Generic     21
System.Concurrency             14
System.Diagnostics             1
System.Disposables             8
System.IO                      1
System.Joins                   34
System.Linq                    13
System.Reactive.Testing        4
System.Reactive.Testing.Mocks  10
System.Threading.Tasks         1
System.Web                     1
System.Windows.Forms           1
System.Windows.Threading       1

C:\Program Files\Microsoft Reactive Extensions SDK\v1.0.10425\Binaries\.NETFramework\v4.0
Microsoft.Reactive.Testing     7
System                         1
System.Reactive                10
System.Reactive.Concurrency    16
System.Reactive.Disposables    9
System.Reactive.Joins          34
System.Reactive.Linq           8
System.Reactive.Subjects       8
System.Reactive.Threading.Tasks 1

Summary of impact

Here is a quick list of the changes that have affected the code from the samples off this blog

Unit has been moved to System.Reactive. I imagine to prevent any conflicts with FSharp’s Unit?

EventLoopScheduler no longer has a constructor that takes a string for the name of the thread, it instead takes a function to act as a Thread Factory so you can use that to specify a name for the thread instead.

Similar methods have been collapsed to overloads. For example; BufferWithTime, BufferWithCount & BufferTimeOrCount are all just Buffer with various overloads. Same goes for WindowWithTime, WIndowWithCount & WindowTimeOrCount. GenerateWithTime is now just an overload of Generate. CreateWithDisposable is now just an overload of Create.

Potentially confusing names have been adjusted. I must admit, I found AsyncSubject and odd name, and also found it odd that Prune was the method you would use to transform an existing observable to one that exhibited AsynchSubject behaviour. The new version of this is the far more appropriately named TakeLast(int). Obviously just pass a value of 1 to get the old semantics of Prune.

The Run method has now been renamed the more obvious (well to new comers) ForEach. It is a bit of an enumerable construct, but that is cool.

My big problem has been the disappearance of the IsEmpty extension method. So I had a long think about this and have decided that you can replace it by creating your own extension method that looks like this

public static class ObservableExtensions
{
  //Is missing now the System.Interactive has been dropped
  public static IObservable<bool> IsEmpty<T>(this IObservable<T> source)
  {
    return source.Materialize()
        .Take(1)
        .Select(n => n.Kind == NotificationKind.OnCompleted
                      || n.Kind == NotificationKind.OnError);
  }
}

For even more details, check out the forum post that has the summary release notes and allsorts of rants about what has changed and what is causing problems for other (+ some helpful tips).

http://social.msdn.microsoft.com/Forums/en-US/rx/thread/527002a3-18af-4eda-8e35-760ca0006b98

This change is not one that you can just take and expect all to be well. However assuming that you can just update some references and add the missing extensions to you own library, you could be fine.

Changes to TestScheduler

The real problems come from the big changes to the testing part of Rx. This appears to have undergone a complete change and most of my existing code does not work. This looks like a big and possibly welcome change to the TestScheduler.

Problems with the previous TestScheduler

There were some fundamental problems with the TestScheduler as it was. There were problems with running it to a specific point in time if there were not any actions scheduled for that point in time. For example, the old TestScheduler would allow you to request that it run to say 5seconds. If there were no actions scheduled, then the clock would not actually advance. You could then Schedule an action to happen at 3Seconds (effectively in the past) and then call Run and it would execute the actions. Hmmm.

TestScheduler scheduler = new TestScheduler();
scheduler.Schedule(()=>{Console.WriteLine("Running at 1seconds");}, TimeSpan.FromSeconds(1));
scheduler.RunTo( scheduler.FromTimeSpan(TimeSpan.FromSeconds(5)));
//The next line is scheduled in the past and somehow this all works!
scheduler.Schedule(()=>{Console.WriteLine("Running at 3seconds");}, TimeSpan.FromSeconds(3));
scheduler.Run();

There also seemed to be a lack of functionality for running to a relative point in time. I have used these extension methods to work around this shortcoming.

public static class TestSchedulerExtensions
{
  /// <summary>
  /// Runs the scheduler from now to the given TimeSpan. Advances relative to it's <c>Now</c> value.
  /// </summary>
  public static void RunNext(this TestScheduler scheduler, TimeSpan interval)
  {
    var tickInterval = scheduler.FromTimeSpan(interval);
    scheduler.RunTo(scheduler.Ticks + tickInterval + 1);
  }

  public static void RunTo(this TestScheduler scheduler, TimeSpan interval)
  {
    var tickInterval = scheduler.FromTimeSpan(interval);
    scheduler.RunTo(tickInterval);
  }

  public static void Step(this TestScheduler scheduler)
  {
    scheduler.RunTo(scheduler.Ticks + 1);
  }

  /// <summary>
  /// Provides a fluent interface so that you can write<c>7.Seconds()</c> instead of <c>TimeSpan.FromSeconds(7)</c>.
  /// </summary>
  /// <param name="seconds">A number of seconds</param>
  /// <returns>Returns a System.TimeSpan to represents the specified number of seconds.</returns>
  public static TimeSpan Seconds(this int seconds)
  {
    return TimeSpan.FromSeconds(seconds);
  }
}

However on initial inspection of the new TestScheduler there appears to be features that support running to an absolute or relative time. It looks like that the Post on testing Rx will need a re-write

Technorati Tags: Reactive-Extensions,Rx

Rxx - Extension to the Reactive Extensions…

2011-05-29T00:00:00.000+01:00

For those of you that use the Rx forums you will have no doubt found an answer or even had your question answered by either James Miles or Dave Sexton. These two have put their mighty brains together to produce some handy extensions to Rx. They went with the fairly obvious (but tongue-in-cheek) name Rxx.

http://social.msdn.microsoft.com/Forums/hu-HU/rx/thread/32f6ee34-5edf-4038-894c-ab47fc893a78

The key features that anyone that has been using Rx for a while will immediately be interested in are the

Tracing/Logging
PropertyChanged and PropertyDescriptor extensions
FileSystemWatcher
and the uber-funky OperationalObservable eg. var os = xs + ys – zs;

For a more complete list look at the documentation.

This could well be a community contribution that could help guide the actual Rx implementations that start coming out of Microsoft (like the Community Extensions around for P&P propjects). James and Dave seem to be taking this quite seriously. They have been checking regularly (with sensible comments!), constructed the codeplex site very well and have already got their issue trackers running.

This is one to keep an eye on.

http://rxx.codeplex.com/

Technorati Tags: Reactive-Extensions,Rx

Rx code from Perth Presentation

2011-05-25T23:50:00.001+01:00

Sorry about the delay in getting this code up. For those who could not make it, my part of the presentation did a bit of an intro and then discussed the testability of Rx and the easy way to deal with streaming data such as pricing in a financial industry.

The key samples from my half of the presentation that raised some interest was the testability of the Photo Gallery View model. The Gallery ViewModel was effectively this

/// <summary>
/// Tested Rx implementation of the ViewModel
/// </summary>
public sealed class RxPhotoGalleryViewModel : INotifyPropertyChanged
{
    public RxPhotoGalleryViewModel(IImageService imageService, ISchedulerProvider scheduler)
    {
        IsLoading = true;
        var files = imageService.EnumerateImages()
                                .ToObservable();

        files
            .SubscribeOn(scheduler.ThreadPool)
            .ObserveOn(scheduler.Dispatcher)
            .Subscribe(
                imagePath =>
                {
                    Images.Add(imagePath);
                },
                () =>
                {
                    IsLoading = false;
                });
    }

    private readonly ObservableCollection<string> _images = new ObservableCollection<string>();
    public ObservableCollection<string> Images
    {
        get { return _images; }
    }

    private bool _isLoading;
    public bool IsLoading
    {
        get { return _isLoading; }
        set
        {
            if (_isLoading != value)
            {
                _isLoading = value;
                InvokePropertyChanged("IsLoading");
            }
        }
    }

    #region Implementation of INotifyPropertyChanged

    public event PropertyChangedEventHandler PropertyChanged;

    public void InvokePropertyChanged(string propertyName)
    {
        PropertyChangedEventHandler handler = PropertyChanged;
        if (handler != null) handler(this, new PropertyChangedEventArgs(propertyName));
    }

    #endregion
}

Except from the constructor, there is really just two properties that expose change notification for the WPF binding engine. Now the code in the constructor is demo-quality in the sense that it is not good practice to do so much work in the constructor. Maybe this would be better

public sealed class RxPhotoGalleryViewModel
{
    private readonly IImageService _imageService;
    private readonly ISchedulerProvider _scheduler;

    public RxPhotoGalleryViewModel(IImageService imageService, ISchedulerProvider scheduler)
    {
        _imageService = imageService;
        _scheduler = scheduler;
    }

    public void Start()
    {
        IsLoading = true;
        var files = _imageService.EnumerateImages()
                                .ToObservable();

        files
            .SubscribeOn(_scheduler.ThreadPool)
            .ObserveOn(_scheduler.Dispatcher)
            .Subscribe(
                imagePath => Images.Add(imagePath),
                () =>IsLoading = false);
    }
//....
}

The test fixture is fairly simple. We pass in a mock implementation of the IImageService and the TestSchedulerProvider similar to the one shown in Rx Part 8 – Testing Rx.

[TestClass]
public class RxPhotoGalleryViewModelTests
{
    private Mock<IImageService> _imageSrvMock;
    private TestSchedulderProvider _testSchedulderProvider;
    private List<string> _expectedImages;

    [TestInitialize]
    public void SetUp()
    {
        _imageSrvMock = new Mock<IImageService>();
        _testSchedulderProvider = new TestSchedulderProvider();

        _expectedImages = new List<string> { "one.jpg", "two.jpg", "three.jpg" };
        _imageSrvMock.Setup(svc => svc.EnumerateImages())
                        .Returns(_expectedImages);

    }

    [TestMethod]
    public void Should_add_ImagesServiceResults_to_Images()
    {
        //Arrange
        // done in setup

        //Act
        var sut = new RxPhotoGalleryViewModel(_imageSrvMock.Object, _testSchedulderProvider);
        _testSchedulderProvider.ThreadPool.Run();
        _testSchedulderProvider.Dispatcher.Run();

        //Assert
        CollectionAssert.AreEqual(_expectedImages, sut.Images);
    }

    [TestMethod]
    public void Should_set_IsLoading_to_true()
    {
        //Arrange
        // done in setup

        //Act
        var sut = new RxPhotoGalleryViewModel(_imageSrvMock.Object, _testSchedulderProvider);
            
        //--NOTE-- note the missing TestScheduler.Run() calls. This will stop any observable being processed. Cool.

        //Assert
        Assert.IsTrue(sut.IsLoading);
    }

    [TestMethod]
    public void Should_set_IsLoading_to_false_when_completed_loading()
    {
        //Arrange
        // done in setup

        //Act
        var sut = new RxPhotoGalleryViewModel(_imageSrvMock.Object, _testSchedulderProvider);
        _testSchedulderProvider.ThreadPool.Run();
        _testSchedulderProvider.Dispatcher.Run();

        //Assert
        Assert.IsFalse(sut.IsLoading);
    }
}

As we now control the scheduling/concurrency we don't have to try to do anything fancy with Dispatchers, BackgroundWorkers, ThreadPools or Tasks which are very difficult to perform unit testing on. Check out the pain that I went through to test responsive WPF apps in this post on Testing Responsive WPF complete with DispatcherFrame and Thread.Sleep(300) in my tests

If you want the running code you can either pull the code down via SVN by following this http://code.google.com/p/rx-samples/source/checkout or you can download the zip.

The PowerPoint presentation is here

You may also be interested in the Design Guidelines produced by the Rx team at Microsoft and also where to get the latest version of Rx

Back to the contents page for Reactive Extensions for .NET Introduction

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Rx session in Perth, Australia

2011-05-11T05:48:00.003+01:00

James Miles and I will be giving a presentation on Rx to a Perth audience this Thursday. If you are in town then come on down!

As those that read this blog will know, Rx is my thing at the moment so feel free to bring plenty of questions.

Full details can be found at the Perth .NET Community of Practice.

Introduction to Rx with Lee Campbell and James Miles

See you there.
Lee

Rx Part 9–Join, Window, Buffer and Group Join

2011-03-14T07:25:00.001Z

While this series of posts is labelled as an introduction, this post takes us past the 100 level style posts. This post looks to tackle the interesting way of combing multiple streams of data. We have looked at versions of combining streams in earlier posts with SelectMany, Merge, Zip, Concat etc. The operators that we look at in this post are different for several reasons

They are new as of 2011
Their matching is based on events coinciding with each other based on some given window of time
They offer some pretty powerful stuff that would otherwise be complex to code and just nasty if you went sans-Rx.

Buffer

Buffer is not a new operator to us, however it can now be conceptually grouped with the group-join operators. These operators all do something with a stream and a window of time. Each operator will open a window when the source stream produces a value. The way the window is closed and which values are exposed is the main difference between each of the operators. Let us first go back to our old friend BufferWithCount that we saw in the second post in this series.

BufferWithCount will create a window when the first value is produced. It will then put that value in to an internal cache. The window will stay open until the count of values has been reached. Each of these values will have been cached. Now that the count has been reached the window will close and the cache will be OnNext’ed as an IList<T>. When the next value is produced from the source, the cache is cleared and we start again. This means that BufferWithCount will take an IObservable<T> and return an IObservable<IList<T>>.

Example Buffer with count of 3

source|-0-1-2-3-4-5-6-7-8-9|
result|-----0-----3-----6-9|
            1     4     7
            2     5     8

In this marble diagram, I have represented the list of values being returned at a point in time as a column of data. i.e. the values 0, 1 & 2 are all returned in the first buffer.

Understanding this it is not much of a leap to understand BufferWithTime. Instead of passing a count we pass a TimeSpan. The closing of the window (and therefore the buffer’s cache) is now dictated by time instead of the count of values produced. This is ever-so more complicated as we now have introduced some sort of scheduling. To produce the IList<T> at the correct point in time we need a scheduler assigned to performing the timing. This also makes testing this stuff a lot easier.

Example Buffer with time of 5 units

source|-0-1-2-3-4-5-6-7-8-9-|
result|----0----2----5----7-|
           1    3    6    8
                4         9

Window

The Window operators are very similar to the Buffer operators, they only really differ by their return type. Where Buffer would take an IObservable<T> and return an IObservable<IList<T>>, the Window operators return an IObservable<IObservable<T>>. It is also worth noting that the Buffer operators will not yield their buffers until the window closes.

Example of Window with a count of 3

source |-0-1-2-3-4-5-6-7-8-9|
window0|-0-1-2|
window1        3-4-5|
window2              6-7-8|
window3                    9|

Example of Window with time of 5 units

source |-0-1-2-3-4-5-6-7-8-9|
window0|-0-1-|
window1      2-3-4|
window2           -5-6-|
window3                7-8-9|

So the obvious difference here is that with the Window operators you get hold of the values from the source as soon as they are produced, but the Buffer operators you must wait until the window closes before the values are accessible.

Switch is the Anti Window

I think it is worth noting, at least from an academic point, that the Window operators produce IObservable<IObservable<T>> and that the Switch operator takes an IObservable<IObservable<T>> and returns an IObservable<T>. As the Window operators ensure that the windows (child streams) do not overlap, we can use the Switch operator to turn a windowed stream back into its original stream.

//is the same as Observable.Interval(TimeSpan.FromMilliseconds(200)).Take(10)
var switchedWindow = Observable.Interval(TimeSpan.FromMilliseconds(200)).Take(10)
    .WindowWithTime(TimeSpan.FromMilliseconds(500))
    .Switch();

Join

Join is not a new Method to the Rx library, but overload we are interested today in is new. From what I can see on the 2 original overloads that take an Array or an IEnumerable of Plan<T>, the usage can be replicated with Merge and Select. They are a bit of a mystery to me.

The overload we are interested in is

public static IObservable<TResult> Join<TLeft, TRight, TLeftDuration, TRightDuration, TResult>
(
    this IObservable<TLeft> left, 
    IObservable<TRight> right, 
    Func<TLeft, IObservable<TLeftDuration>> leftDurationSelector, 
    Func<TRight, IObservable<TRightDuration>> rightDurationSelector, 
    Func<TLeft, TRight, TResult> resultSelector
)

Now this is a fairly hairy signature to try and understand in one go, so let’s take it one parameter at a time.

IObservable<TLeft> left is source stream that defines when a window starts. This is just like the Buffer and Window operators, except that every value published from this source opens a new window. In Buffer and Window, some values just fell into an existing window.

I like to think IObservable<TRight> right as the window value stream. While the left stream controls opening the windows, the right stream will try to pair up with a value from the left stream.

Let us imagine that our left stream produces a value, which creates a new window. If the right stream produces a value while the window is open then the resultSelector function is called with the two values. This is the crux of join, pairing two values from a stream that occur with in the same window. This then leads us to our next question; When does the window close? The answer to this question is both the power and complexity of the Join operator.

When left produces a value, a window is opened. That value is also then passed to the leftDurationSelector function. The result of this function is an IObservable<TLeftDuration>. When that IObservable OnNext’s or Completes then the window for that value is closed. Note that it is irrelevant what the type of TLeftDuration is. This initially left me with the feeling that IObservable<TLeftDuration> was all a bit over kill as you effectively just need some sort of event to say “Closed!”. However by allowing you to use IObservable you can do some clever stuff as we will see later.

So let us first imagine a scenario where we have the left stream producing values twice as fast as the right stream. Imagine that we also never close the windows. We could do this by always returning Observable.Never<Unit>() from the leftDurationSelector function. This would result in the following pairs being produced.

left  |-0-1-2-3-4-5|
right |---A---B---C|

result|---0---0---0
          A   B   C

          1   1   1
          A   B   C

              2   2
              B   C

              3   3
              B   C

                  4
                  C

                  5
                  C

As you can see the left values are cached and replayed each time the right produces a value.

Now it seems fairly obvious that if I immediately closed the window by returning Observable.Empty<Unit> or perhaps Observable.Return(0) that windows would never be opened so no pairs would ever get produced. However what could I do to make sure that these windows did not overlap so that once a second value was produced I would no longer see the first value? Well, if we returned the left stream from the leftDurationSelector that could do it. But wait, when we return the left from the leftDurationSelector it would try to create another subscription and that may introduce side effects. The quick answer to that is to Publish and RefCount the left stream. If we do that the results look more like this.

left  |-0-1-2-3-4-5|
right |---A---B---C|
result|---1---3---5
          A   B   C

This made me think that I could use Join to produce my own version of CombineLatest that we saw in the 5th post in the series. If I had the values from left expire when the next value from left was OnNext’ed then I would be well on my way. However I need the same thing to happen for the right. Luckily the Join operator also provides us with a rightDurationSelector that works just like the leftDurationSelector. This is simple to implement, all I need to do is return a reference to the same left stream when a left value is produced and then the same for the right. The code looks like this.

public static IObservable<TResult> MyCombineLatest<TLeft, TRight, TResult>
(
    IObservable<TLeft> left, 
    IObservable<TRight> right, 
    Func<TLeft, TRight, TResult> resultSelector)
{
    var refcountedLeft = left.Publish().RefCount();
    var refcountedRight = right.Publish().RefCount();
    return Observable.Join(
            refcountedLeft,
            refcountedRight,
            value => refcountedLeft,
            value => refcountedRight,
            resultSelector);
}

While the code above is not production quality (it would need to have some gates in place to mitigate race conditions), it shows us the power that we could get with Join; we can actually use it to create other operators!

GroupJoin

When the Join operator pairs up values that coincide within a window, it would always produce just the left value and the right value to the resultSelector. The GroupJoin operator takes this one step further by passing the left value immediately to the resultSelector with an IObservable of the right values that occur within the window. It’s signature is very similar to Join but note the difference in the resultSelector Func.

public static IObservable<TResult> GroupJoin<TLeft, TRight, TLeftDuration, TRightDuration, TResult>
(
    this IObservable<TLeft> left, 
    IObservable<TRight> right, 
    Func<TLeft, IObservable<TLeftDuration>> leftDurationSelector, 
    Func<TRight, IObservable<TRightDuration>> rightDurationSelector, 
    Func<TLeft, IObservable<TRight>, TResult> resultSelector
)

If we went back to our first Join example where we had

the left producing values twice as fast as the right,
the left never expiring
the right immediately expiring

this is what the result may look like

left              |-0-1-2-3-4-5|
right             |---A---B---C|

0th window values   --A---B---C|
1st window values     A---B---C|
2nd window values       --B---C|
3rd window values         B---C|
4th window values           --C|
5th window values             C|

Now we could switch it around and have it that the left expired immediately and the right never expired the result may look like this

left              |-0-1-2-3-4-5|
right             |---A---B---C|

0th window values   |
1st window values     A|
2nd window values       A|
3rd window values         AB|
4th window values           AB|
5th window values             ABC|

This starts to make things interesting. Sharp readers may have noticed that with GroupJoin you could effectively re-create your own Join by doing something like this

public IObservable<TResult> MyJoin<TLeft, TRight, TLeftDuration, TRightDuration, TResult>(
    IObservable<TLeft> left,
    IObservable<TRight> right,
    Func<TLeft, IObservable<TLeftDuration>> leftDurationSelector,
    Func<TRight, IObservable<TRightDuration>> rightDurationSelector,
    Func<TLeft, TRight, TResult> resultSelector)
{
    return Observable.GroupJoin
        (
            left, 
            right,
            leftDurationSelector,
            rightDurationSelector,
            (leftValue, rightValues)=>rightValues.Select(rightValue=>resultSelector(leftValue, rightValue))
        )
        .Merge();
}

I even was able to knock up my own version of WindowWithTime with this code below

public IObservable<IObservable<T>> MyWindowWithTime<T>(IObservable<T> source, TimeSpan windowPeriod)
{
    return Observable.CreateWithDisposable<IObservable<T>>(o =>
        {
            var windower = new Subject<long>();
            var intervals = Observable.Concat(
                    Observable.Return(0l),
                    Observable.Interval(windowPeriod)
                )
                .Publish()
                .RefCount();

            var subscription = Observable.GroupJoin
                (
                    windower,
                    source.Do(_ => { }, windower.OnCompleted),
                    _ => windower,
                    _ => Observable.Empty<Unit>(),
                    (left, sourceValues) => sourceValues
                )
                .Subscribe(o);
            var intervalSubscription = intervals.Subscribe(windower);
            return new CompositeDisposable(subscription, intervalSubscription);
        });
}

Yeah it is not so pretty, but it is an academic exercise to show case GroupJoin. Those that have read Bart DeSmet’s excellent MiniLinq post (and follow up video) can see that GroupJoin could almost be added to the 3 basic operators Cata, Ana and Bind.

GroupJoin and the other window operators can make otherwise fiddly and difficult tasks a cinch to put together. For example, those in the Finance game can now pretty much use GroupJoin to create their own VWAP and TWAP extension methods. Nice!

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Back to the contents page for Reactive Extensions for .NET Introduction

Back to the previous post; Rx Part 8 - Testing Rx

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Changes to the Rx API

2011-02-20T17:07:00.001Z

Readers of the Reactive Extensions series on this blog will have noticed recently that not all of the code always works* if you download the Rx assemblies and then copy the code off the blog. This is due to the quite frequent changes to the API. I personally have 7 different versions of the libraries and I don’t think I have downloaded them all as they have been released! Also the blog post were started all the way back in May 2010 so it is fair that there has been some movement in the API

*If you get the code from the http://code.google.com/p/rx-samples/source/checkout or as a zip file then it will work, as the correct version of the assembly is included.

For fun I thought I would try to exercise my LINQ skills and write a quick diff tool so I can see what on the public API is actually changing on me. I threw this class together in LinqPad

public class AssemblyDiff
{
    private readonly string _oldAssemblyPath;
    private readonly string _newAssemblyPath;
    
    public AssemblyDiff(string oldAssemblyPath, string newAssemblyPath)
    {
        _oldAssemblyPath = oldAssemblyPath;
        _newAssemblyPath = newAssemblyPath;
    }
    
    public IEnumerable<String> NewMethodNames()
    {
        return MethodNameDelta(GetDeclaredMethods(_newAssemblyPath), GetDeclaredMethods(_oldAssemblyPath));
    }
    
    public IEnumerable<String> DeprecatedMethodNames()
    {
        return MethodNameDelta(GetDeclaredMethods(_oldAssemblyPath), GetDeclaredMethods(_newAssemblyPath));
    }
    
    public static IEnumerable<String> MethodNameDelta(IEnumerable<MethodInfo> original, IEnumerable<MethodInfo> modified)
    {
        return from methodName in original.Select(MethodName).Except(modified.Select(MethodName))
                    orderby methodName
                    select methodName;
    }
    public IEnumerable<MethodInfo> NewMethods()
    {
        var oldMethods = GetDeclaredMethods(_oldAssemblyPath);
        var currentMethods = GetDeclaredMethods(_newAssemblyPath);
        
        return MethodDelta(oldMethods, currentMethods);
    }
    
    public IEnumerable<MethodInfo> DeprecatedMethods()
    {
        var oldMethods = GetDeclaredMethods(_oldAssemblyPath);
        var currentMethods = GetDeclaredMethods(_newAssemblyPath);
        
        return MethodDelta(currentMethods, oldMethods);
    }
    
    public static IEnumerable<MethodInfo> MethodDelta(IEnumerable<MethodInfo> original, IEnumerable<MethodInfo> changed)
    {
        var existingTypes = original.Select(m => m.ReflectedType.FullName)
                                        .Distinct()
                                        .ToList();
        
        return from method in changed.Except(original, new MethodSignatureComparer())
                where existingTypes.Contains(method.ReflectedType.FullName)
                orderby method.ReflectedType.Name, method.Name
                select method;
    }
    
    public IEnumerable<Type> NewTypes()
    {
        var currentTypes = GetTypes(_newAssemblyPath);
        var oldTypes = GetTypes(_oldAssemblyPath);
        
        return from type in currentTypes
                where !oldTypes.Select (t => t.FullName).Contains(type.FullName)
                select type;
    }
    
    public IEnumerable<Type> DeprecatedTypes()
    {
        var currentTypes = GetTypes(_newAssemblyPath);
        var oldTypes = GetTypes(_oldAssemblyPath);
        
        return from type in oldTypes
                where !currentTypes.Select (t => t.FullName).Contains(type.FullName)
                select type;
    }
    
    private static IEnumerable<MethodInfo> GetAllMethods(string path)
    {
        return  from type in GetTypes(path)
                from method in type.GetMethods()
                where method.IsPublic
                select method;
    }
    
    private static IEnumerable<MethodInfo> GetDeclaredMethods(string path)
    {
        return GetAllMethods(path).Where(method => method.DeclaringType == method.ReflectedType);
    }
    
    private static IEnumerable<Type> GetTypes(string path)
    {
        return  from file in Directory.EnumerateFiles(path, "*.dll")
                from module in Assembly.LoadFrom(file).GetModules()
                from type in module.GetTypes()
                where type.IsPublic
                select type;
    }
    
    private static string MethodName(MethodInfo m)
    {
        return string.Format("{0}.{1}", m.ReflectedType.Name, m.Name);
    }
    
    public static string MethodSignature(MethodInfo m)
    {
        //return m.ToString();
        var ps = m.GetParameters();
        var args = ps.Select(p=>ParameterSignature(p, ps.Length));
        var argsDemlimted = string.Join(",", args);
        
        return string.Format("{0} {1}.{2}({3})", m.ReturnType.Name, m.ReflectedType.Name, m.Name, argsDemlimted);
    }
        
    private static string ParameterSignature(ParameterInfo parameter, int parameterCount)
    {
        var modifier = "";//out/ref/params/
        var defaultValue = "";
        if(parameter.IsOut) modifier = "out ";
        if(parameter.IsOptional)
        {
            modifier = "optional ";
            defaultValue = parameter.DefaultValue.ToString();
        }
        if(parameter.IsRetval) modifier += "isretval ";
        if(parameter.IsIn) modifier += "IsIn ";
        if(parameter.IsLcid) modifier += "IsLcid ";
        if(parameter.Position== parameterCount-1 && parameter.ParameterType.IsArray)
        {
            modifier = "params ";
        }
        return string.Format("{0}{1}{2}", modifier,parameter.Name, defaultValue);
    }
    
    private class MethodSignatureComparer : IEqualityComparer<MethodInfo>
    {
        public bool Equals(MethodInfo lhs, MethodInfo rhs)
        {                
            return string.Equals(lhs.ToString(), rhs.ToString());
        }
        
        public int GetHashCode(MethodInfo method)
        {
            return method.ToString().GetHashCode();
        }
    }
}

and then I used it like this

var old =         @"C:\Program Files\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2787.0\Net4";
var current =     @"C:\Program Files\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2838.0\Net4";

var dllDiff = new AssemblyDiff(old, current);

"New Methods".Dump();
dllDiff.NewMethodNames().Dump();
"Deprecated Methods".Dump();
dllDiff.DeprecatedMethodNames().Dump();
"New Types".Dump();
dllDiff.NewTypes().Select (t => t.FullName).Dump();
"Deprecated Types".Dump();
dllDiff.DeprecatedTypes().Select (t => t.FullName).Dump();
"New overloads".Dump();
dllDiff.NewMethods().Select(AssemblyDiff.MethodSignature).Dump();
"Deprecated overloads".Dump();
dllDiff.DeprecatedMethods().Select(AssemblyDiff.MethodSignature).Dump();

and I get this neat output.

New Methods (7 Items)

ConnectableObservable`2.Connect
ConnectableObservable`2.Subscribe
Observable.GroupJoin
Observable.Multicast
Observable.Window
Qbservable.GroupJoin
Qbservable.Window

Deprecated Methods (6 Items)

ConnectableObservable`1.Connect
ConnectableObservable`1.Subscribe
Observable.Prune
Observable.Replay
Qbservable.Prune
Qbservable.Replay

New Types (1 Item)

System.Collections.Generic.ConnectableObservable`2

Deprecated Types (0 Items)

New Overloads (30 Items)

IEnumerable`1 EnumerableEx.Generate(initialState,condition,iterate,resultSelector)
IObservable`1 Observable.BufferWithTime(source,timeSpan,timeShift,scheduler)
IObservable`1 Observable.BufferWithTime(source,timeSpan,scheduler)
IObservable`1 Observable.BufferWithTime(source,timeSpan,timeShift)
IObservable`1 Observable.BufferWithTime(source,timeSpan)
IObservable`1 Observable.BufferWithTimeOrCount(source,timeSpan,count,scheduler)
IObservable`1 Observable.BufferWithTimeOrCount(source,timeSpan,count)
IObservable`1 Observable.GroupJoin(left,right,leftDurationSelector,rightDurationSelector,resultSelector)
IObservable`1 Observable.If(condition,thenSource)
IObservable`1 Observable.Join(left,right,leftDurationSelector,rightDurationSelector,resultSelector)
IConnectableObservable`1 Observable.Multicast(source,subject)
IObservable`1 Observable.Publish(source,subject)
IObservable`1 Observable.Publish(source,subject,selector)
IObservable`1 Observable.Window(source,windowOpenings,windowClosingSelector)
IObservable`1 Observable.Window(source,windowClosingSelector,scheduler)
IObservable`1 Observable.Window(source,windowClosingSelector)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan,timeShift,scheduler)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan,scheduler)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan,timeShift)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan)
IQbservable`1 Qbservable.BufferWithTimeOrCount(source,timeSpan,count,scheduler)
IQbservable`1 Qbservable.BufferWithTimeOrCount(source,timeSpan,count)
IQbservable`1 Qbservable.GroupJoin(left,right,leftDurationSelector,rightDurationSelector,resultSelector)
IQbservable`1 Qbservable.If(provider,condition,thenSource)
IQbservable`1 Qbservable.Join(left,right,leftDurationSelector,rightDurationSelector,resultSelector)
IQbservable`1 Qbservable.Publish(source,subject,selector)
IQbservable`1 Qbservable.Publish(source,subject)
IQbservable`1 Qbservable.Window(source,windowOpenings,windowClosingSelector)
IQbservable`1 Qbservable.Window(source,windowClosingSelector,scheduler)
IQbservable`1 Qbservable.Window(source,windowClosingSelector)

Deprecated overloads (71 Items)

IEnumerable`1 EnumerableEx.Generate(initialState,condition,resultSelector,iterate)
IEnumerable`1 EnumerableEx.Generate(function)
IEnumerable`1 EnumerableEx.Generate(initialState,resultSelector,iterate)
IEnumerable`1 EnumerableEx.Generate(initial,resultSelector,iterate)
IEnumerable`1 EnumerableEx.Generate(initial,condition,resultSelector,iterate)
IObservable`1 Observable.BufferWithTime(source,timeSpan,timeShift,scheduler)
IObservable`1 Observable.BufferWithTime(source,timeSpan,scheduler)
IObservable`1 Observable.BufferWithTime(source,timeSpan,timeShift)
IObservable`1 Observable.BufferWithTime(source,timeSpan)
IObservable`1 Observable.BufferWithTimeOrCount(source,timeSpan,count,scheduler)
IObservable`1 Observable.BufferWithTimeOrCount(source,timeSpan,count)
IConnectableObservable`1 Observable.Prune(source)
IConnectableObservable`1 Observable.Prune(source,scheduler)
IObservable`1 Observable.Prune(source,selector)
IObservable`1 Observable.Prune(source,selector,scheduler)
IObservable`1 Observable.Publish(source1,source2,selector)
IObservable`1 Observable.Publish(source1,source2,selector,scheduler)
IObservable`1 Observable.Publish(source1,source2,source3,selector)
IObservable`1 Observable.Publish(source1,source2,source3,selector,scheduler)
IObservable`1 Observable.Publish(source1,source2,source3,source4,selector)
IObservable`1 Observable.Publish(source1,source2,source3,source4,selector,scheduler)
IConnectableObservable`1 Observable.Publish(source,initialValue)
IConnectableObservable`1 Observable.Publish(source,initialValue,scheduler)
IObservable`1 Observable.Publish(source,selector,initialValue)
IObservable`1 Observable.Publish(source,selector,initialValue,scheduler)
IConnectableObservable`1 Observable.Publish(source)
IConnectableObservable`1 Observable.Publish(source,scheduler)
IObservable`1 Observable.Publish(source,selector)
IObservable`1 Observable.Publish(source,selector,scheduler)
IConnectableObservable`1 Observable.Replay(source)
IConnectableObservable`1 Observable.Replay(source,scheduler)
IObservable`1 Observable.Replay(source,selector)
IObservable`1 Observable.Replay(source,selector,scheduler)
IConnectableObservable`1 Observable.Replay(source,window)
IObservable`1 Observable.Replay(source,selector,window)
IConnectableObservable`1 Observable.Replay(source,window,scheduler)
IObservable`1 Observable.Replay(source,selector,window,scheduler)
IConnectableObservable`1 Observable.Replay(source,bufferSize,scheduler)
IObservable`1 Observable.Replay(source,selector,bufferSize,scheduler)
IConnectableObservable`1 Observable.Replay(source,bufferSize)
IObservable`1 Observable.Replay(source,selector,bufferSize)
IConnectableObservable`1 Observable.Replay(source,bufferSize,window)
IObservable`1 Observable.Replay(source,selector,bufferSize,window)
IConnectableObservable`1 Observable.Replay(source,bufferSize,window,scheduler)
IObservable`1 Observable.Replay(source,selector,bufferSize,window,scheduler)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan,timeShift,scheduler)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan,scheduler)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan,timeShift)
IQbservable`1 Qbservable.BufferWithTime(source,timeSpan)
IQbservable`1 Qbservable.BufferWithTimeOrCount(source,timeSpan,count,scheduler)
IQbservable`1 Qbservable.BufferWithTimeOrCount(source,timeSpan,count)
IQbservable`1 Qbservable.Prune(source,selector,scheduler)
IQbservable`1 Qbservable.Prune(source,selector)
IQbservable`1 Qbservable.Publish(source,selector,initialValue)
IQbservable`1 Qbservable.Publish(source,selector,scheduler)
IQbservable`1 Qbservable.Publish(source,selector,initialValue,scheduler)
IQbservable`1 Qbservable.Publish(source,selector)
IQbservable`1 Qbservable.Publish(source1,source2,selector)
IQbservable`1 Qbservable.Publish(source1,source2,selector,scheduler)
IQbservable`1 Qbservable.Publish(source1,source2,source3,selector)
IQbservable`1 Qbservable.Publish(source1,source2,source3,selector,scheduler)
IQbservable`1 Qbservable.Publish(source1,source2,source3,source4,selector)
IQbservable`1 Qbservable.Publish(source1,source2,source3,source4,selector,scheduler)
IQbservable`1 Qbservable.Replay(source,selector,bufferSize)
IQbservable`1 Qbservable.Replay(source,selector,bufferSize,window)
IQbservable`1 Qbservable.Replay(source,selector,bufferSize,window,scheduler)
IQbservable`1 Qbservable.Replay(source,selector)
IQbservable`1 Qbservable.Replay(source,selector,scheduler)
IQbservable`1 Qbservable.Replay(source,selector,window)
IQbservable`1 Qbservable.Replay(source,selector,window,scheduler)
IQbservable`1 Qbservable.Replay(source,selector,bufferSize,scheduler)

From memory Generate has been a constant source of change which has confused some readers that are using different versions of the library to what the Part 2 post was done with. This diff script goes to show that it is still undergoing changes

Links:

Reactive Extensions for .NET an Introduction

DevLabs: Reactive Extensions for .NET (Rx)

Silverlight testing

2011-02-13T15:02:00.002Z

I am putting this out there to see if I can get some traction with other Test Driven Silverlight coders out there. If you are one of these people you will know of the strife your day-to-day coding. For those who don't know what I mean these are the three options a Silverlight developer has with regards to test driven coding:

1) Use the Silverlight Unit Test Framework. The problem with this is that you lose any integrated development support, it is amazingly slow (in the area of 1-5 tests per second), and doesn't have any useful build tool support (coverage, TFS, TeamCity). Massive Fail.

2) Cross compile to .NET 4 all of your Models, ViewModels, Controllers, Modules, Presenters (i.e. everything that is not a View or a Control). Now write unit tests against this project. This means you get back to fast tests (100s tests per second) but take a hit on compiling twice and managing the project linking and just having twice as many projects floating around.

3) What I imagine as the most popular option, just don't write any tests.

Looking at what most of the requests are for, tells me most people are using Silverlight for Marketing websites to stream rich content. Business applications have yet to stake any dominance. What I am hoping that anyone reading this will if they feel my pain, just go to this link and vote. It seems that these polls really have an effect; DataTemplates appear to be part of SL 5 due to massive demand. I am hoping that Microsoft can focus on getting the underlying framework right before they go off and give us a 3D-multitouch-proximity aware API :)

http://dotnet.uservoice.com/forums/4325-silverlight-feature-suggestions/suggestions/313397-unit-testing-integrated-in-visual-studio-and-msbui?ref=title

Rx design guidelines

2010-11-02T14:16:00.003Z

The Rx team have released a pdf specifiying their Design Guidelines to use when coding with the Reactive Extentions. The original post is here http://blogs.msdn.com/b/rxteam/archive/2010/10/28/rx-design-guidelines.aspx The PDF is here http://go.microsoft.com/fwlink/?LinkID=205219 Great stuff. Next; the FxCop rules for static analysis?

Rx Part 8 - Testing Rx

2010-10-26T08:38:00.003+01:00

Having reviewed the scheduling available to us in Rx and now that we are aware of Hot and Cold observables we have almost enough skills in our tool belt to start using Rx in anger. However many developers wouldn’t dream of starting coding without first being able to write tests to prove their code is in fact satisfying their requirements and providing them with that safety net against regression. Rx poses some interesting problems to our Test Driven community

Scheduling and therefore Threading are generally avoided in test scenarios as they can introduce race conditions which may lead to non-deterministic tests.
Tests should run as fast as possible.
Rx is a new technology/library so naturally as we master it, we will refactor our code. We want to use to tests to ensure our refactoring have not altered the internal behaviour of our code base.

So we want to test our code but don't want to introduce false-negatives, false-positives or non-deterministic tests. Also if we look at the Rx library there are plenty of methods that involve Scheduling so it is hard to ignore. This Linq query shows us that there are at least 32 extension methods that accept an IScheduler as a parameter

var query = from method in typeof (Observable).GetMethods()
        from parameter in method.GetParameters()
        where typeof (IScheduler).IsAssignableFrom(parameter.ParameterType)
        group method by method.Name into m
        orderby m.Key
        select m.Key;
            
query.Run(Console.WriteLine);
/* 
BufferWithTime, Catch, Concat, Delay, Empty, Generate, GenerateWithTime, Interval
Merge, ObserveOn, OnErrorResumeNext, Prune, Publish, Range, Repeat, Replay
Retry, Return, Sample, Start, StartWith, Subscribe, SubscribeOn, Take, Throttle
Throw, TimeInterval, Timeout, Timer, Timestamp, ToAsync, ToObservable
*/

There are some methods that we already will be familiar with such as ObserveOn and SubscribeOn. Then there are others that will optionally take an IScheduler in one of the method overloads. TDD/TestFirst coders will want to opt for the overload that takes the IScheduler so that we can have some control over scheduling in our tests.

In this example we create a stream that publishes values every second for 5 seconds.

var interval = Observable
    .Interval(TimeSpan.FromSeconds(1))
    .Take(5);

If we to write a test that ensured that we received 5 values and they were each 1 second apart it would take 5 seconds. That would be no good, I want hundreds if not thousands of tests to run in 5 seconds. A more common time related example that would need tests is setting a timeout.

var stream = Observable.Never<int>();

var exceptionThrown = false;
stream.Timeout(TimeSpan.FromMinutes(1))
    .Run(
        i => Console.WriteLine("This will never run."),
        ex => exceptionThrown = true);
Assert.IsTrue(exceptionThrown);

This test would take 1 minute to run. However if we did some test first code in this style, where we added the timeout after the test, running the test initially to fail (Red-Green-Refactor) would never complete. Hmmmm…..

TestScheduler

To our rescue comes the TestScheduler. This is a recent addition to the Rx library that takes the concept of a virtual scheduler to allow us emulate and control time. Cool.

The concept of a virtual scheduler can sort of be thought of a queue of actions to be executed that are each marked with the point in time they should be executed. Where actions are attempted to be scheduled at the same “time” the second collision will be scheduled or queued after the first action but marked as with the same point in “time” . When we use the TestScheduler we can either “drain the queue” by calling run which will execute all scheduled actions, or we can specify to run all tasks up to a point in time.

In this example we schedule a task on to the queue to be run immediately by using the simple overload. We then execute everything scheduled for the first tick (ie the first action).

var scheduler = new TestScheduler();
var wasExecuted = false;
scheduler.Schedule(() => wasExecuted = true);
Assert.IsFalse(wasExecuted);
scheduler.RunTo(1);         //execute 1 tick of queued actions
Assert.IsTrue(wasExecuted);

The TestScheduler type is actually an implementation of the abstract VirtualScheduler<TAbsolute, TRelative> where the TestSchuduler specifies long for both TAbsolute and TRelative. The net result is that the TestScheduler interface looks something like this

public class TestScheduler
{
    public TestScheduler();

    // Methods
    public long FromTimeSpan(TimeSpan timeSpan);
    public long Increment(long absolute, long relative);
    public void Run();
    public void RunTo(long time);
    public IDisposable Schedule(Action action);
    public IDisposable Schedule(Action action, TimeSpan dueTime);
    public IDisposable Schedule(Action action, long dueTime);
    public void Sleep(long ticks);
    public DateTimeOffset ToDateTimeOffset(long absolute);

    // Properties
    public DateTimeOffset Now { get; }
    public long Ticks { get; }
}

We should already be familiar with what the Schedule methods would do, the other method of interest for this post are the Run() and RunTo(long) methods. Run() will just execute all the actions that have been scheduled. RunTo(long) will execute all the actions that have been scheduled up to the time specified by the long value which represents ticks in this case. Having a quick look into the implementations for Schedule, RunTo and Run give us the insight we need to really understand the virtual schedulers.

public IDisposable Schedule(Action action)
{
    return this.Schedule(action, TimeSpan.Zero);
}
public IDisposable Schedule(Action action, TimeSpan dueTime)
{
    return this.Schedule(action, this.FromTimeSpan(dueTime));
}
public IDisposable Schedule(Action action, TRelative dueTime)
{
    BooleanDisposable disposable = new BooleanDisposable();
    TAbsolute local = this.Increment(this.Ticks, dueTime);
    Action action2 = delegate {
        if (!disposable.IsDisposed)
        {
            action();
        }
    };
    ScheduledItem<TAbsolute> item = new ScheduledItem<TAbsolute>(action2, local);
    this.queue.Enqueue(item);
    return disposable;
}

public void RunTo(TAbsolute time)
{
    while ((this.queue.Count > 0) && (this.queue.Peek().DueTime.CompareTo(time) <= 0))
    {
        ScheduledItem<TAbsolute> item = this.queue.Dequeue();
        this.Ticks = item.DueTime;
        item.Action();
    }
}

public void Run()
{
    while (this.queue.Count > 0)
    {
        ScheduledItem<TAbsolute> item = this.queue.Dequeue();
        this.Ticks = item.DueTime;
        item.Action();
    }
}

Obviously this is the internal implementation and would be subject to change, however as the documentation currently is quite weak for the VirtualScheduler and TestScheduler I think this use of reflection is helpful. This example should clear up what happens when items are scheduled at the same time.

var scheduler = new TestScheduler();
long dueTime = 4L;
scheduler.Schedule(() => Console.WriteLine("1"), dueTime);
scheduler.Schedule(() => Console.WriteLine("2"), dueTime);
scheduler.Schedule(() => Console.WriteLine("3"), dueTime+1);
scheduler.Schedule(() => Console.WriteLine("4"), dueTime+1);
Console.WriteLine("RunTo(dueTime)");
scheduler.RunTo(dueTime); 
Console.WriteLine("Run()");
scheduler.Run();
/* Output:
RunTo(dueTime)
1
2
Run()
3
4
*/

Testing Rx code

Now that we have learnt a little bit about the TestScheduler, lets look at how we could use it to get our 2 initial code snippets (Interval and TimeOut) to execute as fast as possible but still maintaining the semantics of time. In this example we generate our 5 values one second apart but pass in our TestScheduler to the Interval method.

[TestMethod]
public void Testing_with_test_scheduler()
{
    var scheduler = new TestScheduler();
    var interval = Observable
        .Interval(TimeSpan.FromSeconds(1), scheduler)
        .Take(5);

    bool isComplete = false;
    interval.Subscribe(Console.WriteLine, () => isComplete = true);

    scheduler.Run();

    Assert.IsTrue(isComplete); //Executes in less than 0.01s "on my machine"
}

While this is mildly interesting, what I think is more important is how we would test a real piece of code. Imagine if you will a Presenter that subscribes to a stream of prices. As prices are published it adds them to a ViewModel. Assuming this is a WPF or Silverlight implementation we take the liberty of enforcing that the subscription be done on the ThreadPool and the observing is executed on the Dispatcher.

public class MyPresenter
{
...
    public void Show(string symbol)
    {
        _myService.PriceStream(symbol)
                                    .SubscribeOn(Scheduler.ThreadPool)
                                    .ObserveOn(Scheduler.Dispatcher)
                                    .Subscribe(price=>_viewModel.Prices.Add(price));
    }
...
}

While this snippet of code may do what we want it to do, it will be hard to test as it is accessing the schedulers via static properties. To help my testing, I have created my own interface that exposes the same IScheduler implementations that the Scheduler type does.

public interface ISchedulerService
{
    IScheduler CurrentThread { get; }
    IScheduler Dispatcher { get; }
    IScheduler Immediate { get; }
    IScheduler NewThread { get; }
    IScheduler ThreadPool { get; }
    //IScheduler TaskPool { get; }
}
public sealed class SchedulerService : ISchedulerService
{
    public IScheduler CurrentThread { get { return Scheduler.CurrentThread; } }

    public IScheduler Dispatcher { get { return Scheduler.Dispatcher; } }

    public IScheduler Immediate { get { return Scheduler.Immediate; } }

    public IScheduler NewThread { get { return Scheduler.NewThread; } }

    public IScheduler ThreadPool { get { return Scheduler.ThreadPool; } }

    //public IScheduler TaskPool { get { return Scheduler.TaskPool; } }
}

This now allows me to inject an ISchedulerService which gives me a seam to help my testing. I can now write some tests for my Presenter.

[TestInitialize]
public void SetUp()
{
    _myServiceMock = new Mock<IMyService>();
    _viewModelMock = new Mock<IViewModel>();
    _schedulerService = new TestSchedulers();

    var prices = new ObservableCollection<decimal>();
    _viewModelMock.SetupGet(vm => vm.Prices).Returns(prices);
    _viewModelMock.SetupProperty(vm => vm.IsConnected);
}

[TestMethod]
public void Should_pass_symbol_to_MyService_PriceStream()
{
    var expected = "SomeSymbol";
    var priceStream = new Subject<decimal>();
    _myServiceMock.Setup(svc => svc.PriceStream(It.Is<string>(symbol=>symbol==expected))).Returns(priceStream);

    var sut = new MyPresenter(_myServiceMock.Object, _viewModelMock.Object, _schedulerService);
    sut.Show(expected);

    _myServiceMock.Verify();
}

[TestMethod]
public void Should_add_to_VM_Prices_when_MyService_publishes_price()
{
    decimal expected = 1.23m;
    var priceStream = new Subject<decimal>();
    _myServiceMock.Setup(svc => svc.PriceStream(It.IsAny<string>())).Returns(priceStream);

    var sut = new MyPresenter(_myServiceMock.Object, _viewModelMock.Object, _schedulerService);
    sut.Show("SomeSymbol");
    _schedulerService.ThreadPool.Schedule(() => priceStream.OnNext(expected));  //Schedule the OnNext
    _schedulerService.ThreadPool.RunTo(1);  //Execute the OnNext action
    _schedulerService.Dispatcher.RunTo(1);  //Execute the OnNext Handler (ie adding to the Prices collection)

    Assert.AreEqual(1, _viewModelMock.Object.Prices.Count);
    Assert.AreEqual(expected, _viewModelMock.Object.Prices.First());
}

These two tests show first a simple expectation that a string value passed to my Show(string) method will be passed to the underlying service. This is not at all relevant to Rx. The next test shows the usage of my implementation of the ISchedulerService specific for testing. It exposes all of the IScheduler properties as instances of TestSchedulers. This now allows me to inject TestSchedulers for testing which in-turn allows me to control the rate at which things are scheduled.

For those of you new to Moq, some of the syntax my be a little bit confusing. Where you see a Setup(..) or SetupGet(..) method call there is just a little bit of Expression magic that tells my mocks to return correct thing when called. The .Object property hanging off my mocks are the dynamically generated implementations of the interfaces. This next test I hope you find the most interesting. Here we really get the full value out of our TestScheduler, by testing timeouts.

[TestMethod]
public void Should_timeout_if_no_prices_for_10_seconds()
{
    var timeoutPeriod = TimeSpan.FromSeconds(10);
    var priceStream = Observable.Never<decimal>();
    _myServiceMock.Setup(svc => svc.PriceStream(It.IsAny<string>())).Returns(priceStream);

    var sut = new MyPresenter(_myServiceMock.Object, _viewModelMock.Object, _schedulerService);
    sut.Show("SomeSymbol");

    _schedulerService.ThreadPool.RunTo(timeoutPeriod.Ticks - 1);
    Assert.IsTrue(_viewModelMock.Object.IsConnected);

    _schedulerService.ThreadPool.RunTo(timeoutPeriod.Ticks);
    Assert.IsFalse(_viewModelMock.Object.IsConnected);
}

The key points to note are:

We return an Observable.Never for our price stream so that no prices are ever pushed and no OnComplete is published either.
The _schedulerService is a test fake that returns TestSchedulers for all of it’s schedulers
We run the ThreadPool TestScheduler up until 1 tick away from our timeout period and ensure that we have not timed out
We run the ThreadPool TestScheduler up to our timeout period and then ensure that we have timed out.
The test is sub second even though we are testing for a 10 second timeout!

Generally I only want to have one assertion in each of my tests, but for this example I think it elegantly tests that the Timeout is for 10seconds and not longer or shorter. If I was to remove the first assertion my test would only prove that the timeout was no greater than 10 seconds but could reasonably be set to say 3 seconds. The implementation for this simple presenter is as follows

public class MyPresenter
{
    private readonly IMyService _myService;
    private readonly IViewModel _viewModel;
    private readonly ISchedulerService _schedulerService;

    public MyPresenter(IMyService myService, IViewModel viewModel, ISchedulerService schedulerService)
    {
        _myService = myService;
        _schedulerService = schedulerService;
        _viewModel = viewModel;
    }

    public void Show(string symbol)
    {
        _myService.PriceStream(symbol)
                    .SubscribeOn(_schedulerService.ThreadPool)
                    .ObserveOn(_schedulerService.Dispatcher)
                    .Timeout(TimeSpan.FromSeconds(10), _schedulerService.ThreadPool)
                    .Subscribe(OnPriceUpdate, ex =>
                                                    {
                                                        if (ex is TimeoutException)
                                                            _viewModel.IsConnected = false;
                                                    });
        _viewModel.IsConnected = true;
    }

    private void OnPriceUpdate(decimal price)
    {
        _viewModel.Prices.Add(price);
    }
}

I hope this post on testing helps you bring you skills that you have been developing in Rx to a level where you may be comfortable considering them for production use.

For your reference here are the two test versions of the ISchedulerService I find useful. One will just schedule everything with the ImmediateScheduler and the other uses the TestSchedulers for fine grained control.

public sealed class TestSchedulers : ISchedulerService
{
    private readonly TestScheduler _currentThread = new TestScheduler();
    private readonly TestScheduler _dispatcher = new TestScheduler();
    private readonly TestScheduler _immediate = new TestScheduler();
    private readonly TestScheduler _newThread = new TestScheduler();
    private readonly TestScheduler _threadPool = new TestScheduler();

    #region Implementation of ISchedulerService
    IScheduler ISchedulerService.CurrentThread { get { return _currentThread; } }

    IScheduler ISchedulerService.Dispatcher { get { return _dispatcher; } }

    IScheduler ISchedulerService.Immediate { get { return _immediate; } }

    IScheduler ISchedulerService.NewThread { get { return _newThread; } }

    IScheduler ISchedulerService.ThreadPool { get { return _threadPool; } }
    #endregion

    public TestScheduler CurrentThread { get { return _currentThread; } }

    public TestScheduler Dispatcher { get { return _dispatcher; } }

    public TestScheduler Immediate { get { return _immediate; } }

    public TestScheduler NewThread { get { return _newThread; } }

    public TestScheduler ThreadPool { get { return _threadPool; } }
}

public sealed class ImmediateSchedulers : ISchedulerService
{
    public IScheduler CurrentThread { get { return Scheduler.Immediate; } }

    public IScheduler Dispatcher { get { return Scheduler.Immediate; } }

    public IScheduler Immediate { get { return Scheduler.Immediate; } }

    public IScheduler NewThread { get { return Scheduler.Immediate; } }

    public IScheduler ThreadPool { get { return Scheduler.Immediate; } }
}

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Back to the contents page for Reactive Extensions for .NET Introduction

Back to the previous post; Rx Part 7 – Hot and Cold Observables

Forward to next post; Part 9 – Join, Window, Buffer and Group Join

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Reactive Extensions for .NET an Introduction

2010-08-24T17:38:00.002+01:00

Welcome to my series introducing Reactive Extensions to .NET (Rx). This series is aimed at any .NET developer curious about the IObservable<T> and IObserver<T> interfaces that have popped up in .NET 4. However Rx is a library so, ~~Silverlight 3~~ & Silverlight 4 developers as well as .NET 3.5, Windows Phone and JavaScript coders can download the library. Rx is big. It is big in allsorts of ways:

In the way that it tackles the Observer pattern is bold
In the way it tackles concurrency is quite a shift from how I have done it before.
The number of (extension) methods is huge.
The way in which it integrates with LINQ to leverage LINQ's composability & declarative style
The fact that any .NET developer should care. UI developer, backend algorithm coder or integrator; It helps all of us.
The future plans are even more grand, but that is a different series all together :-)

In this series I will introduce you to:

the new types the Rx will provide
the extension methods and how to use them
how to manage subscriptions to "streams" of data
how to deal with concurrency to your advantage and avoid the common old pitfalls
how to compose, aggregate and transform streams
how to build workflows with Rx
some tips and tricks I have picked while using Rx over the past months.

So download the assemblies to reference, fire up Visual Studio and let's get started:

The full source code is now available either via SVN at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Edit – This series of posts was first created in mid 2010. The Rx framework has gone through numerous changes during and after the writing of these posts. I am constantly making an effort to ensure that the blog stays as current as I can make it –Lee Campbell

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Rx Part 7 – Hot and Cold Observables

2010-08-19T12:31:00.001+01:00

In this post we will look how to describe and handle 2 styles of observable streams

Streams that are passive and start publishing on request,
Streams that are active and publish regardless of subscriptions.

In this sense passive streams are called Cold and active are described as being Hot. You can draw some similarities between implementations of the IObservable<T> interface and implementations of the IEnumerable<T> interface with regards to Hot and Cold. With IEnumerable<T> you could have an “On demand” collection via the yield return syntax, or you could have an eager evaluation by populating a List<T>, for example, and returning that (as per the example below)

Do(LazyEvaluation());
Do(EagerEvaluation());


private void Do(IEnumerable<int> list)
{
  foreach (var i in list)
  {
    Console.WriteLine("Read out first value of {0}", i);
    break;
  }
}

public IEnumerable<int> LazyEvaluation()
{
  Console.WriteLine("About to return 1");
  yield return 1;
  Console.WriteLine("About to return 2");//Never called in this example
  yield return 2;
}

public IEnumerable<int> EagerEvaluation()
{
  var result = new List<int>();
  Console.WriteLine("About to return 1");
  result.Add(1);
  Console.WriteLine("About to return 2");//executed but not used.
  result.Add(2);
  return result;
}

Implementations of IObservable<T> can exhibit similar variations in style.

Examples of Hot observables that could publish regardless of if there are any subscribers would be:

Mouse movements
Timer events
broadcasts like ESB channels or UDP network packets.
price ticks from a trading exchange

Some examples of Cold observables would be:

subscription to a queue
when Rx is used for an asynchronous request
on demand streams

In this post we will look at 3 scenarios in which cold, hot and both cold & hot are implemented.

Cold Observables

In this first example we have a requirement to fetch a list of products from a service. In our implementation we choose to return an IObservable<string> and as we get the results we publish them until we have the full list and then we publish an OnComplete. This is a pretty simple example.

private static IObservable<string> GetProducts()
{
  return Observable.CreateWithDisposable<string>(
    o =>
    {
      using(var conn = new SqlConnection(@"Data Source=.\SQLSERVER;Initial Catalog=AdventureWorksLT2008;Integrated Security=SSPI;"))
      using (var cmd = new SqlCommand("Select Name FROM SalesLT.ProductModel", conn))
      {
        conn.Open();
        SqlDataReader reader = cmd.ExecuteReader(CommandBehavior.CloseConnection);
        while (reader.Read())
        {
          o.OnNext(reader.GetString(0));
        }
        o.OnCompleted();
        return Disposable.Create(()=>Console.WriteLine("--Disposed--"));
      }
    });
}

This style of API would allow for a non blocking call to fetch the list of products and would inform the consumer of when the list was complete. This is fairly common stuff, but note that every time this is called, the database will be accessed again.

In the example above I use Disposable.Create factory method. This factory method just creates an implementation of IDisposable that executes a given action when disposed. This is perfect for doing a Console.WriteLine once the subscription has been disposed.

In this example below, we have a consumer of our above code, but it explicitly only wants up to 3 values (the full set has 128 values). This code illustrates that the Take(3) expression will restrict what the consumer receives but GetProducts() method will still publish all of the values.

public void ColdSample()
{
  var productStream = GetProducts().Take(3);
  productStream.Subscribe(Console.WriteLine);
  Console.ReadLine();
}

Hot Observables

Trying to come up with an example for Hot Observables has been a real pain. I have started off with examples with some sort of context (streaming stock prices or weather information) but this all seemed to detract from the real working of the code. So I think it is best to step through this slowly with a contrived demo and build it up to a piece of code you might actually want to use.

Let us start with subscribing to an Interval. In the example below we subscribe to the same Observable that is created via the Interval extension method. The delay between the two subscriptions should demonstrate that while they are subscribed to the same observable instance, it is not the same logical stream of data.

public void SimpleColdSample()
{
  var period = TimeSpan.FromSeconds(1);
  var observable = Observable.Interval(period);
  observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i));
  Thread.Sleep(period);
  observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i));
  Console.ReadKey();
  /* Ouput:
   first subscription : 0
   first subscription : 1
   second subscription : 0
   first subscription : 2
   second subscription : 1
   first subscription : 3
   second subscription : 2   
   */
}

Publish and Connect

If I want to be able to share the actual stream of data and not just the instance of the observable, I can use the Publish() extension method. This will return an IConnectableObservable<T>, which extends IObservable<T> by adding the single Connect() method. By using the Publish() then the Connect() method, we can get this functionality.

public void SimpleConnectSample()
{
  var period = TimeSpan.FromSeconds(1);
  var observable = Observable.Interval(period).Publish();
  observable.Connect();
  observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i));
  Thread.Sleep(period);
  observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i));
  Console.ReadKey();
  /* Ouput:
   first subscription : 0
   first subscription : 1
   second subscription : 1
   first subscription : 2
   second subscription : 2   
   */
}

In the example above the observable variable is an IConnectableObservable<T>, and by calling Connect() it will subscribe to the underlying (the Observable.Interval). In this case we are quick enough to subscribe before the first item is published but only on the first subscription. The second subscription subscribes late and misses the first publication. We could move the invocation of the Connect() method until after each of the subscriptions have been made so that even with the Thread.Sleep we wont really subscribe to the underlying until after both subscriptions are made. This would be done as follows:

public void SimpleConnectSample()
{
  var period = TimeSpan.FromSeconds(1);
  var observable = Observable.Interval(period).Publish();
  observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i));
  Thread.Sleep(period);
  observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i));
  observable.Connect();
  Console.ReadKey();
  /* Ouput:
   first subscription : 0
   second subscription : 0
   first subscription : 1
   second subscription : 1
   first subscription : 2
   second subscription : 2   
   */
}

You can probably imagine how this could be quite useful where an application had the need to share streams of data. In a trading application if you wanted to consume a price stream for a certain asset in more than one place, you would want to reuse that stream and not have to make another subscription to the server providing that data. Publish() and Connect() are perfect solutions for this.

Disposal of connections and subscriptions

What does become interesting is how disposal is performed. What was not covered above is that the Connect() method returns an IDisposable. By disposing of the “connection” you can turn the stream on and off (Connect() to turn it on and then disposing of the connection to turn it off). In this example we see that the the stream can be connected and disconnected multiple times.

public void ConnectAndDisposeSample()
{
  var period = TimeSpan.FromSeconds(1);
  var observable = Observable.Interval(period).Publish();
  observable.Subscribe(i => Console.WriteLine("subscription : {0}", i));
  var exit = false;
  while (!exit)
  {
    Console.WriteLine("Press enter to connect, esc to exit.");
    var key = Console.ReadKey(true);
    if(key.Key== ConsoleKey.Enter)
    {
      var connection = observable.Connect();  //--Connects here--
      Console.WriteLine("Press any key to dispose of connection.");
      Console.ReadKey();
      connection.Dispose();                   //--Disconnects here--
    }
    if(key.Key==ConsoleKey.Escape)
    {
      exit = true;
    }
  }
  /* Ouput:
   Press enter to connect, esc to exit.
   Press any key to dispose of connection.
   subscription : 0
   subscription : 1
   subscription : 2
   Press enter to connect, esc to exit.
   Press any key to dispose of connection.
   subscription : 0
   subscription : 1
   subscription : 2
   Press enter to connect, esc to exit.   
   */
}

Let us finally consider automatic disposal of a connection. It would be common place for a single stream to be shared between subscriptions, as per the price stream example mentioned above. It would however also be common place for the developer to want to only have the stream running hot if there are subscriptions to it. Therefore it seems not only obvious that there should be a mechanism for automatically connecting (once a subscription has been made), but also a mechanism for disconnecting (once there are no more subscriptions) from a stream. First let us look at what happens to a stream when we connect with no subscribers, and then later unsubscribe:

public void OrphanedStreamExample()
{
  var period = TimeSpan.FromSeconds(1);
  var observable = Observable.Interval(period)
    .Do(l => Console.WriteLine("Publishing {0}", l)) //produce Side effect to show it is running.
    .Publish();
  observable.Connect();
  Console.WriteLine("Press any key to subscribe");
  Console.ReadKey();
  var subscription = observable.Subscribe(i => Console.WriteLine("subscription : {0}", i));

  Console.WriteLine("Press any key to unsubscribe.");
  Console.ReadKey();
  subscription.Dispose();

  Console.WriteLine("Press any key to exit.");
  Console.ReadKey();
  /* Ouput:
   Press any key to subscribe
   Publishing 0
   Publishing 1
   Press any key to unsubscribe.
   Publishing 2
   subscription : 2
   Publishing 3
   subscription : 3
   Press any key to exit.
   Publishing 4
   Publishing 5
   */
}

A few things to note here:

I use the Do extension method to create side effects on the stream (ie writing to the console). This allows us to see when the stream is actually connected.
We connect first and then subscribe, which means we can be publishing without any subscriptions.
We dispose of our subscription but don’t dispose of the connection which means the stream will still be running. This means we will be publishing even though there are no subscriptions.

RefCount

Taking the last example, if we just comment out the line that makes the Connection, and then add a further extension method to our creation of our observable RefCount we have magically implemented all of our requirements. RefCount will take an IConnectableObservable<T> and turn it back into an IObservable<T> and automatically implement the connect and disconnect behaviour we are looking for.

public void RefCountExample()
{
  var period = TimeSpan.FromSeconds(1);
  var observable = Observable.Interval(period)
    .Do(l => Console.WriteLine("Publishing {0}", l)) //produce Side effect to show it is running.
    .Publish()
    .RefCount();
  //observable.Connect(); Use RefCount instead now
  Console.WriteLine("Press any key to subscribe");
  Console.ReadKey();
  var subscription = observable.Subscribe(i => Console.WriteLine("subscription : {0}", i));

  Console.WriteLine("Press any key to unsubscribe.");
  Console.ReadKey();
  subscription.Dispose();

  Console.WriteLine("Press any key to exit.");
  Console.ReadKey();
  /* Ouput:
   Press any key to subscribe
   Press any key to unsubscribe.
   Publishing 0
   subscription : 0
   Publishing 1
   subscription : 1
   Publishing 2
   subscription : 2
   Press any key to exit.
   */
}

Other Connectable Observables

While this is a post about Hot and Cold Observables, I think it is worth mentioning the other ways IConnectableObservable<T> can pop up.

Prune

The prune method is effectively a non blocking .Last() call. You can consider it similar to an AsyncSubject<T> wrapping your target Observable so that you get equivalent semantics of only returning the last value of an observable and only once it completes.

public void PruneExample()
{
  var period = TimeSpan.FromSeconds(1);
  var observable = Observable.Interval(period)
    .Take(5)
    .Do(l => Console.WriteLine("Publishing {0}", l)) //produce Side effect to show it is running.
    .Prune();
  observable.Connect();
  Console.WriteLine("Press any key to subscribe");
  Console.ReadKey();
  var subscription = observable.Subscribe(i => Console.WriteLine("subscription : {0}", i));

  Console.WriteLine("Press any key to unsubscribe.");
  Console.ReadKey();
  subscription.Dispose();

  Console.WriteLine("Press any key to exit.");
  Console.ReadKey();
  /* Ouput:
   Press any key to subscribe
   Publishing 0
   Publishing 1
   Press any key to unsubscribe.
   Publishing 2
   Publishing 3
   Publishing 4
   subscription : 4
   Press any key to exit.
   */
}

Replay

The Replay extension method allows you take an existing Observable and give it “replay” semantics as per the ReplaySubject<T>. As a reminder, the ReplaySubject<T> will cache all values so that any late subscribers will also get all of the values. In this example 2 subscriptions are made on time, and then a third subscription can be made after they complete. Even though the third subscription can be done after the OnComplete we can still get all of the values.

public void ReplayOnHotExample()
{
  var period = TimeSpan.FromSeconds(1);
  var hot = Observable.Interval(period)
    .Take(3)
    .Publish();
  hot.Connect();
  Thread.Sleep(period); //Run hot and ensure a value is lost.
  var observable = hot.Replay();
  observable.Connect();
  observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i));
  Thread.Sleep(period);
  observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i));

  Console.ReadKey();
  observable.Subscribe(i => Console.WriteLine("third subscription : {0}", i));
  Console.ReadKey();

  /* Ouput:
   first subscription : 1
   second subscription : 1
   first subscription : 2
   second subscription : 2   
   third subscription : 1
   third subscription : 2
   */
}

I hope that gives some insight to Hot and Cold observables. I think this is one of the more mysterious parts of Rx for many newbies so I hope that it helps clear up some of the cool stuff you can do with Rx. I sure many of you will look forward to finding ways to implement this in your next application.

For more information on IConnectableObservable<T> and Hot/Cold streams check out these resources:

channel 9 video on Hot and Cold observables

Hot and Cold by Bnaya Eshet

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Back to the contents page for Reactive Extensions for .NET Introduction

Back to the previous post; Part 6 - Scheduling and threading

Forward to next post; Part 8 – Testing Rx

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Rx Part 6 – Scheduling and Threading

2010-06-27T23:08:00.003+01:00

So far in the series of posts we have managed to avoid any explicit usage of threading or concurrency. There are some methods that we have covered that implicitly will be introducing some level of concurrency to perform their jobs (e.g. : Buffer, Delay, Sample etc all require a separate thread to do their magic). However most of this has been kindly abstracted away from us. This post will look at the beauty of the Rx API and its ability to effectively remove the need for WaitHandles, and any explicit calls to using Threads, the ThreadPool and the new shiny Task type.

A friend of mine once wisely stated that you should always understand at least one layer below what you are coding. At the time he was referring to networking protocols, but I think it is sage advice for all programming. On the current project I am working on there are some very savvy developers that are very comfortable working in a multithreaded environment. The project has client and server side threading problems that we have had to tackle. I believe the whole team would agree that it has bee amazing that amount of concurrency that Rx will handle for you in a declarative way. The code base is virtually free of WaitHandles, Monitor or lock usage, or any explicit creation of threads. This has evolved into this state over time as we have come to grips with the power of Rx and the end result is far cleaner code. However, having the experience on the team allowed us to find out ways we should and shouldn’t be using Rx which would have been just too hard for me to do alone.

Getting back to my friend’s comment about understanding the underlying subsystem, this is especially important when dealing with Rx and scheduling. Just because Rx abstracts some of this away, it does not mean that you cant still create problems for yourself if you are not careful. Before I scare you too much let’s look at some of the Scheduling features of Rx.

Scheduling

In the Rx world, you can control the scheduling of two things

The invocation of the subscription
The publishing of notifications

As you could probably guess these are exposed via two extension methods to IObservable<T> called SubscribeOn and ObserveOn. Both methods have an overload that take an IScheduler and will return an IObservable<T> so you can chain methods together.

public static class Observable
{
  public static IObservable<TSource> ObserveOn<TSource>(
      this IObservable<TSource> source, IScheduler scheduler) 
  {...}
public static IObservable<TSource> SubscribeOn<TSource>(
      this IObservable<TSource> source, IScheduler scheduler) 
  {...}
}

public interface IScheduler
{
    IDisposable Schedule(Action action);
    IDisposable Schedule(Action action, TimeSpan dueTime);
    DateTimeOffset Now { get; }
}

The IScheduler interface is of less interest to me than the types that implement the interface. Depending on your platform* (Silverlight3, Silverlight4, .Net 3.5, .Net 4.0) you will be exposed appropriate implementations via a static class Scheduler. These are the static properties that you can find on the Scheduler type that expose different schedulers.

Scheduler.Dispatcher will ensure that the actions are performed on the Dispatcher, which is obviously useful for Silverlight and WPF applications. You can imagine that the implementation for this would just delegate any calls to ISchedule(Action) straight to Dispatcher.BeginInvoke(Action)

Scheduler.NewThread will schedule all actions onto a new thread.

Scheduler.ThreadPool will schedule all actions onto the Thread Pool.

Scheduler.TaskPool (which is only available to Silverlight 4 and .NET 4.0) will schedule actions onto the TaskPool.

Scheduler.Immediate will ensure the action is not scheduled but is executed immediately.

Scheduler.CurrentThread just ensures that the actions are performed on the thread that made the original call. This is different to Immediate, as CurrentThread will queue the action to be performed. Note the difference in the output of the following code. One method passes Scheduler.Immediate, the other passes Scheduler.CurrentThread.

private static void ScheduleTasks(IScheduler scheduler)
{
    Action leafAction = () => Console.WriteLine("leafAction.");
    Action innerAction = () =>
                             {
                                 Console.WriteLine("innerAction start.");
                                 scheduler.Schedule(leafAction);
                                 Console.WriteLine("innerAction end.");
                             };
    Action outerAction = () =>
                             {
                                 Console.WriteLine("outer start.");
                                 scheduler.Schedule(innerAction);
                                 Console.WriteLine("outer end.");
                             };
    scheduler.Schedule(outerAction);
}

public void CurrentThreadExample()
{
    ScheduleTasks(Scheduler.CurrentThread);
    Console.ReadLine();
    /*Output:
     * outer start.
     * outer end.
     * innerAction start.
     * innerAction end.
     * leafAction.
     */
}

public void ImmediateExample()
{
    ScheduleTasks(Scheduler.Immediate);
    Console.ReadLine();
    /*Output:
     * outer start.
     * innerAction start.
     * leafAction.
     * innerAction end.
     * outer end.
     */
}

*Sorry Rx for JavaScript, I have not even opened the box on you and don’t know anything about scheduling in JavaScript.

Examples

So they are each of our Schedulers, lets see some of them in use. The think I want to point out here is that the first few times I used these overloads I had them confused as to what they actually did. You should use the SubscribeOn method to describe how you want any warm up and background processing code to be scheduled. ObserveOn method is used to describe where you want your notification scheduled to. So for example, if you had a WPF application that used Rx to populate and ObservableCollection<T> then you would almost certainly want to use SubscribeOn with one of the Threaded schedulers (NewThread, ThreadPool or maybe TaskPool) and then you would have to use the Dispatcher scheduler to update your collection.

public void LoadCustomers()
{
    _customerService.GetCustomers()
        .SubscribeOn(Scheduler.NewThread)
        .ObserveOn(Scheduler.Dispatcher)
        .Subscribe(Customers.Add);
}

So all of the schedulers just offer a nice abstraction to us to utilise the various ways we can write concurrent code. Besides saving me from having to write the tedious code to get code onto a new thread or thread pool it also makes Rx threading easy. Oh Rx, you thought I had forgotten. I didn’t think that any of the schedulers except Current & Immediate warranted a further explanation but, I do think it is worth pointing out some of the “fun” threading problems you can face even though the scheduling has been abstracted away from you.

Deadlocks

When writing the current application my team is working on we found out the hard way that Rx code can most certainly deadlock. When you consider that some calls (like .First() ) are blocking, and that we can schedule work to be done in the future, it becomes obvious that race condition can apply. This example is the most simple deadlock I could think of. It is fairly silly but it will get the ball rolling.

var stream = new Subject<int>();
Console.WriteLine("Next line should deadlock the system.");
var value = stream.First();
stream.OnNext(1);
Console.WriteLine("I can never execute....");

Hopefully we wont ever write code that silly, and if we did our tests would give us fairly quick feed back that things were wrong. What lets deadlocks slip into the system is when they manifest themselves at integration points. This example may be a little harder to find but is only small step away from the silly 1st example. Here we block in the constructor on a UI element which will always be created on the dispatcher. The blocking call is waiting for an event, that can only be raised from the dispatcher – deadlock.

public Window1()
{
    InitializeComponent();
    DataContext = this;
    Value = "Default value";

    //Deadlock! We need the dispatcher to continue
    // to allow me to click the button to produce a value.
    Value = _subject.First(); 

    //This will give same result but will not be blocking(deadlocking).
    _subject.Take(1).Subscribe(value => Value = value);
}

private void MyButton_Click(object sender, RoutedEventArgs e)
{
    _subject.OnNext("New Value");
}

public string Value
{
    get { return _value; }
    set
    {
        _value = value;
        var handler = PropertyChanged;
        if (handler != null) handler(this, new PropertyChangedEventArgs("Value"));
    }
}

In this example we start seeing things that can become more sinister. This example has a Button that the click command will try to get the first value from an Observable exposed via an interface.

public partial class Window1 : INotifyPropertyChanged
{
    private readonly IMyService _service = new MyService(); //Imagine DI here.
    private int _value2;

    public Window1()
    {
        InitializeComponent();
        DataContext = this;
    }

    public int Value2
    {
        get { return _value2; }
        set
        {
            _value2 = value;
            var handler = PropertyChanged;
            if (handler != null) handler(this, new PropertyChangedEventArgs("Value2"));
        }
    }

    #region INotifyPropertyChanged Members

    public event PropertyChangedEventHandler PropertyChanged;

    #endregion

    private void MyButton2_Click(object sender, RoutedEventArgs e)
    {
        Value2 = _service.GetTemperature().First();
    }
}

There is only a small problem here in that we block on the Dispatcher thread (.First() is a blocking call), however this manifest's itself into a deadlock if the service code is written incorrectly.

class MyService : IMyService
{
    public IObservable<int> GetTemperature()
    {
        return Observable.Create<int>(
            o =>
                {
                    o.OnNext(27);
                    o.OnNext(26);
                    o.OnNext(24);
                    return () => { };
                })
            .SubscribeOnDispatcher();
    }
}

This odd implementation with explicit scheduling will cause the 3 OnNext calls to be scheduled once the .First() call has finished, which is waiting for an OnNext to be called – Deadlock.

So far this post has been a bit doom and gloom about scheduling and the problems you could face, that is not the intent. I just wanted to make it obvious that Rx was not going to solve the age old concurrency problems, but it will make it easier to get it right if you follow this simple rule.

Only the final subscriber should be setting the scheduling.
Avoid using .First() –Ed: that is for you Olivier. We will cal this rule 1b

Where the last example came unstuck is that the service was dictating the scheduling paradigm when really it had no business doing so. Before we had a clear idea of where we should be doing the scheduling in my current project, we had allsorts of layers adding “helpful” scheduling code. What it ended up creating was a threading nightmare. When we removed all scheduling code and then located it in a single layer (at least in the Silverlight client) most of our concurrency problems went away. I recommend you do the same. At least in WPF/Silverlight applications, the pattern should be simple: “Subscribe on a Background thread; Observe on the Dispatcher”.

So my challenge to the readers is to add to the comments:

Any other scheduling rules (2 seems quite small, and I was only going to have 1)
Post some nasty Rx race condition code
What rules do you have for Subscribing on the background thread? Which Scheduler should I use and when i.e. NewThread, ThreadPool & TaskPool. – and I come full circle about understanding one layer below that to which you are working.

RX Part 5 – Combining multiple IObservable streams

2010-06-19T17:04:00.001+01:00

In the last post we covered some of the flow control features of Rx and how to conceptualise them with Marble diagrams. This post will continue to build on those concepts by looking at different ways of working with multiple streams.

The Concat extension method is probably the most simple extension method. If you have covered the previous flow control post then most of the error handling constructs are more complex than this method. The method will simple publish values from the second stream once the first stream completes.

//Generate values 0,1,2
var stream1 = Observable.Generate(0, i => i < 3, i => i, i => i + 1);
//Generate values 100,101,102,103,104
var stream2 = Observable.Generate(100, i => i < 105, i => i, i => i + 1);

stream1
    .Concat(stream2)
    .Subscribe(Console.WriteLine);
Console.ReadLine();
/* Returns:
 * 
 * stream1 --0--0--0--|
 * stream2 -----------0--0--0--0--|
 * 
 * result  --0--0--0--0--0--0--0--|
 */

If either stream was to OnError then the result stream would OnError too. This means that if stream1 produced an OnError then stream2 would never be used. If you wanted stream2 to be used regardless of if stream1 produced an OnError or not then the extension method OnErrorResumeNext would be your best option.

Quick Video on Concat, Catch and OnErrorResume next on Channel9.

The Amb method was a new concept to me. I believe this comes from functional programming and is an abbreviation of Ambiguous. Effectively this extension method will produce values from the stream that first produces values and will completely ignore the other stream. In the examples below I have 2 streams that both produce values. In the first example stream1 will win the race and the result stream will be stream1’s values. In the second example, I delay the stream1 from producing values so stream2 will win the race and the result stream will be the values from stream2.

//Generate values 0,1,2
var stream1 = Observable.Range(0,3);
//Generate values 100,101,102,103,104
var stream2 = Observable.Range(100,5);

stream1
    .Amb(stream2)
    .Subscribe(Console.WriteLine);
Console.ReadLine();
/* Returns:
 *  if stream 1 produces a value first...
 * stream1 --0--0--0--|
 * stream2 ---0--0--0--0--0--|
 * 
 * result  --0--0--0--|     //All from stream1
 */

stream1.Delay(TimeSpan.FromMilliseconds(100))
    .Amb(stream2)
    .Subscribe(Console.WriteLine);
Console.ReadLine();
/* Returns:
 * stream1 ---0--0--0--|
 * stream2 --0--0--0--0--0--|
 * 
 * result  --0--0--0--0--0--|     //All from stream2
 */

The Merge extension method does a primitive combination of multiple streams where they implement the same type of T. The result will also be an IObservable<T> but will have the values produced to the result stream as the occur in the source streams. The stream will complete when all of the source streams complete or when an OnError is published by any stream.

//Generate values 0,1,2
var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3);
//Generate values 100,101,102,103,104
var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(5).Select(i => i + 100);
stream1
    .Merge(stream2)
    .Subscribe(Console.WriteLine);
Console.ReadLine();
/*
 * Returns:
 * stream1 ----0----0----0|
 * stream2 --0--0--0--0--0|
 * 
 * result  --0-00--00-0--00-|  
 * Output:
 * 100
 * 0
 * 101
 * 102
 * 1
 * 103
 * 104      //Note this is a race condition. 2 could be
 * 2        //  published before 104.
 */

Merge also provides other overloads that allow you to pass more than 2 source observables via an IEnumerable or params arrays. The Overload that take a params array it great for when we know how many streams we want to merge at compile time, and the IEnumerable overload is better for when we dont know at compile time how many streams we need to merge.

//Create a third stream
var stream3 = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(10).Select(i => i + 200);

//Number of streams known at compile time.
Observable.Merge(stream1, stream2, stream3)
    .Subscribe(Console.WriteLine);
Console.ReadLine();

//We can dynamically create a list at run time with this overload.
var streams = new List<IObservable<long>>();
streams.Add(stream1);
streams.Add(stream2);
streams.Add(stream3);
Observable.Merge(streams).Subscribe(Console.WriteLine);
Console.ReadLine();

A quick video on Merge on Channe9.

SelectMany, like it’s counter part in IEnumerable<T> extension method will create the Caretisan product of the two streams. So for every item in one stream, it will give you every item in the other stream. A primitive way to think of it is a nexted for loop that creates a 2D array. If you want more info on SelectMany I will leave it to you to do a google search as this fairly well documented in the IEnumerable world.

//Generate values 0,1,2
var stream1 = Enumerable.Range(0, 3).ToObservable();
//Generate values 100,101,102,103,104
var stream2 = Enumerable.Range(100, 5).ToObservable();
stream1
    .SelectMany(i => stream2, (lhs, rhs) => new { Left = lhs, Right = rhs })
    .Subscribe(Console.WriteLine);
Console.ReadLine();
/*
 * Output.
 * { Left = 0, Right = 100 }
 * { Left = 0, Right = 101 }
 * { Left = 1, Right = 100 }
 * { Left = 0, Right = 102 }
 * { Left = 1, Right = 101 }
 * { Left = 2, Right = 100 }
 * { Left = 0, Right = 103 }
 * { Left = 1, Right = 102 }
 * { Left = 2, Right = 101 }
 * { Left = 0, Right = 104 }
 * { Left = 1, Right = 103 }
 * { Left = 2, Right = 102 }
 * { Left = 1, Right = 104 }
 * { Left = 2, Right = 103 }
 * { Left = 2, Right = 104 }
 */

A quick Video on SelectMany on channel9

Zip is another interesting merge feature. Just like a Zipper on clothing or a bag, the Zip method will bring together two sets of values as pairs; two-by-two. Things to note about the Zip function is that the result stream will complete when the first of the streams complete, it will error if either of the streams error and it will only publish once it was a pair. So if one of the source streams publish values faster than the other stream, the rate of publishing will be dictated by the slower of the two streams.

//Generate values 0,1,2
var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3);
//Generate values a,b,c,d,e,f
var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(6).Select(i => Char.ConvertFromUtf32((int)i + 97));
stream1
    .Zip(stream2, (lhs, rhs) => new { Left = lhs, Right = rhs })
    .Subscribe(Console.WriteLine);
Console.ReadLine();
/* Returns:
 * stream1 ----0----1----2|        stream 1 values represented as ints
 * stream2 --a--b--c--d--e--f|   s2 values represented as chars
 * 
 * result  ----0----1----2|
 *             a    b    c
 */

Here are two short videos on Zip (first, second)on Channel9. Note the second video is actually incorrect, can you spot why?

CombineLatest is worth comparing to the zip method. Both methods will use a function that takes a value from each stream to produce the result value. The difference is that CombineLatest will cache the last value of each stream, and when either stream produces a new value then that new value and that last value from the other stream will be sent to the result function. This example uses the same inputs as the previous Zip example but note that many more values are produced. The leaves CombineLatest somewhere between Zip and SelectMany :-)

//Generate values 0,1,2
var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3);
//Generate values a,b,c,d,e,f
var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(6).Select(i => Char.ConvertFromUtf32((int)i + 97));
stream1
    .CombineLatest(stream2, (s1, s2) => new { Left = s1, Right = s2 })
    .Subscribe(Console.WriteLine, () => Console.WriteLine("Complete"));
Console.ReadLine();
/*
 * Returns:
 * stream1 ----0----1----2|        stream 1 values represented as ints
 * stream2 --a--b--c--d--e--f|     stream 2 values represented as chars
 * 
 * result  ----00--01-1--22-2|     the result as pairs.
 *             ab  cc d  de f    
 */

Quick video on CombineLatest on Channel9

ForkJoin like the last few extension methods also requires a function to produce the result but this will only return the last values from each stream. Things to note with ForkJoin is that like the previous methods, if either stream error so will the result stream, but if either stream is empty (ie completes with no values) then the result stream will also be empty. This example uses the same values as the previous samples and will only produce a pair from the last values from each stream once they both complete.

//Generate values 0,1,2
var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3);
//Generate values a,b,c,d,e
var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(5).Select(i => Char.ConvertFromUtf32((int)i + 97));
stream1
    .ForkJoin(stream2, (s1, s2) => new { Left = s1, Right = s2 })
    .Subscribe(Console.WriteLine);
Console.ReadLine();
/*
 * Returns:
 * stream1 ----0----1----2|        stream 1 values represented as ints
 * stream2 --a--b--c--d--e|   s2 values represented as chars
 * 
 * result  --------------2|   the result as pairs. 
 *                       e     
 */

One thing to note with most of the extension methods discussed is that they generally have a matching static method that takes a params array or IEnumerable as discussed in the Merge chapter.

So having looked at these ways to bring multiple observables together we have implicitly brought some concurrency and threading to our code. This allows us to nicely lead into the next post in the series which will be on Scheduling and Threading with Rx.

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Back to the contents page for Reactive Extensions for .NET Introduction

Back to the previous post; Part 4 - Flow control

Forward to next post; Part 6 - Scheduling and threading

Technorati Tags: Reactive-Extensions,Rx,Tutorial

RX part 4 – Flow control

2010-05-29T11:37:00.001+01:00

In the previous post we looked at the lifecycle of a subscription to a stream. A subscription can be terminated by unsubscribing/disposing and the stream (IObservable<T>) can terminate due to it completing naturally via an OnCompleted or erroneously with an OnError. OnError creates an interesting problem; it publishes the Exception, it does not throw it. This means that you can not use the standard Structured Exception Handling process of try/catch with Observables. Well that is not entirely true. In this example the stream raises an OnError that we can catch with a try/catch block thanks to the extension method overload of subscribe I use. It will take an OnError and then throw the Exception.

try
{
    var subject = new Subject<int>();
    subject.Subscribe(Console.WriteLine);
    //Or explicitly rethrow as per this line.
    //subject.Subscribe(Console.WriteLine, ex => { throw ex.PrepareForRethrow(); });
    subject.OnNext(1);
    subject.OnError(new IOException("test exception"));
}
catch (IOException ioEx)
{
    Console.WriteLine("Caught IO exception: {0}", ioEx.Message);
}

Throwing published OnError exceptions does not allow for a very composable style of coding. It may be useful in cases like the example above, but the power of Rx as you will see in future posts is the ability to create compositions of streams together. Once we are running compositions of streams then SEH is not very helpful or useful. Rx provides several methods to provide a composition friendly way to handle errors and exceptions.

Visualising streams

Before I cover the flow control methods that Rx offers I want to divert quickly and talk about a visual tool we will use to help communicate the concepts relating to streams. Marble diagrams are great for communicating Rx streams and you may find them useful for describing any stream, except for the completely basic. When using marble diagrams to communicate Rx streams there are only a few things you need to know

a stream is represented by a horizontal line
time moves to the right (ie things on the left happened before things on the right)
we only need 3 symbols to represent an Event
1. “0” for OnNext
2. “X” for an OnError
3. “|” for OnCompleted

This is a sample of a stream that publishes 3 values and then completes

--0--0--0—|

This is a sample of a stream that publishes 4 values then errors.

--0--0--0--0--X

While these examples may seem too simple to warrant a visual representation, the simplicity of marble diagrams are great once we using multiple streams.

Flow control constructs

Sometimes when dealing with an observable, it is conceivable that errors may occur that are acceptable and we should try again. Imagine that we want this effect where the error in stream 1 (S1) is acceptable, we try again on stream 2 (S2). The last line is composition of the two streams that is the result we want to expose (R)

S1--0--0--X
S2-----------0--0--0--0
R --0--0-----0--0--0--0

In the example above we could recreate this with several methods.

Retry is the most simple method available to us. Retry will try to re-subscribe to the IObservable<T> on any failure. In this example we just use the simple overload that will always retry on any exception.

public static void RetrySample<T>(IObservable<T> stream)
{
    stream.Retry().Subscribe(t=>Console.WriteLine(t)); //Will always retry the stream
    Console.ReadKey();
}
/*
Given stream that will produce 0,1,2 then error; the output would be
0
1
2
0
1
2
0
1
2
.....
*/

which would look like this as a marble diagram

S--0--0--0--x--0--0--0--x--0--0--0--
R--0--0--0-----0--0--0-----0--0--0--

Alternatively we can specify the max number of times to retry. In this example we only retry once, therefore the error that gets published on the second subscription will be passed up to the final subscription. Note that to retry once you pass a value of 2. Maybe the method should be called Try?

public static void RetryOnceSample<T>(IObservable<T> stream)
{
    stream.Retry(2)
        .Subscribe(t=>Console.WriteLine(t), 
		 ex=>Console.WriteLine("Gave up on 2nd Error")); 
    Console.ReadKey();
}
/*Ouput:
0
1
2
0
1
2
Gave up on 2nd Error
*/

As a marble diagram this would look like

S--0--0--0--x--0--0--0--x
R--0--0--0-----0--0--0--x

OnErrorResumeNext may cause some old VB developers to shudder but it offers a different route to use than Retry. While retry will always try to re-subscribe to the same stream; OnErrorResumeNext takes another IObservable<T> as a parameter to use when the original stream publishes and error. In this example when the stream1 publishes and error we re-subscribe to stream2.

public static void FailoverSample<T>(IObservable<T> stream, IObservable<T> failover)
{
    stream
        .OnErrorResumeNext(failover)
        .Subscribe(t=>Console.WriteLine(t), ex => Console.WriteLine(ex));
}
/*
stream  --"1"--"2"--"3"--X
failover--------------------"a"--"b"--"c"--X
result  --"1"--"2"--"3"-----"a"--"b"--"c"--|
*/

An important thing to note here is that when the second stream publishes an error the result stream just completes and ignores the error.

Catch is probably the most useful method to use as the previous 2 methods will react the same regardless of the type of exception published. Catch however allows you to specify which exceptions it can catch just like a normal catch block. This example is similar to the last example, but we will explicitly state the exception we should catch that would allow us to failover to the next stream.

public static void CatchSample<T>(IObservable<T> stream, IObservable<T> failover)
{
    stream
        .Catch((InvalidOperationException ex) => failover)
        .Subscribe(t => Console.WriteLine(t), 
                   ex => Console.WriteLine(ex), 
                   () => Console.WriteLine("Completed"));
}

Catch has other overloads that allow you to pass a params array of Observable<T> or an Enumerable<Observable<T>> instead of just specifying an intial and a failover stream. This means you can effectively have a large list of streams to try when a previous one fails. Of course if any of them actually complete then the next one will not be used. If the last one publishes an OnError then that error will then be published to the IObservalbe that the Catch method returned.

Another pair of interesting methods is the Materialize and Dematerialize methods. Materialize will flatten an observable’s three different publication types (OnNext, OnError & OnCompleted) into wrapped publications of a Notifcation<T> type. Notification<T> is an abstract class that exposes 4 properties and an Overloaded Accept method

public abstract class Notification<T> : IEquatable<Notification<T>>
{
    // Properties
    public abstract Exception Exception { get; }
    public abstract bool HasValue { get; }
    public abstract NotificationKind Kind { get; }
    public abstract T Value { get; }

    // Methods
    public abstract TResult Accept<TResult>(IObserver<T, TResult> observer);
    public abstract void Accept(IObserver<T> observer);
    public abstract void Accept(Action<T> onNext, Action<Exception> onError, Action onCompleted);
    public abstract TResult Accept<TResult>(Func<T, TResult> onNext, Func<Exception, TResult> onError, Func<TResult> onCompleted);

    // ...
}

By inspecting the Kind property you can identify which type of publication you have received. If the Kind is OnError you can access its published Exception via the Exception property. Accessing the Value property will get you either

the OnNext value,
throw the OnError exception
or throw an InvalidOperationException if the Kind is OnCompleted

The HasValue property just provides the convenience to check if the Kind is OnNext so you can safely get the value without having an exception thrown.

As you can imagine the Materialise method can be useful for Logging the content of a stream. In this example we create an extension method that logs all events to the Console.

public static class ExampleExtensions
{
    /// <summary>
    /// Logs implicit notifications to console.
    /// </summary>
    /// <example>
    /// <code>myStream.Log().Subscribe(....);</code>
    /// </example>
    public static IObservable<T> Log<T>(this IObservable<T> stream)
    {
        return stream.Materialize()
        .Do(n => Console.WriteLine(n))
        .Dematerialize();
    }
}

Note that here we use Dematerialize to take our stream of Notification<T> and transform it back to our original stream. You could also use Materialize to create your own more powerful Catch methods, but we will look at applications of Rx later.

Wire tapping a stream

The Do extension method was used in the last example and it would not be fair to continue with out explaining what it does. Do method is used to provide side effects upon a stream. In the example above the side-effect is that we wrote to the console. This is different to a Subscribe because the Do method returns you an IObservable<T> which can be thought of as the same IObservable<T> that was passed to it, however the Subscribe method returns you an IDisposable. I like to think of the Do method as a wire tap to a stream. ;-)

It would be unfair to mention Do and leave out Run. The Run method is very similar to the Do method except for two things:

Run returns void
Run is a blocking call
Run can be called without any parameters. This effectively is a call to block until OnCompleted or OnError is published.
Run has no overload that takes an Action for the OnCompleted publication. It doesn't make sense to do so as the method will just stop blocking when OnCompleted is published so you can just invoke the action you would other wise pass to OnCompleted immediately after the Run method returns.

The things in common with the Run and Do method overloads is they both provide an Overload that:

takes an Action<T> to be performed on each OnNext publication.
takes an Action<T> for OnNext and an Action<Exception> for OnError publications
takes an IObserver<T> that will be used to handle the publications explicitly

Now that we have had a brief look at Flow control and aggregating streams together, in the next post we will uncover the other aggregation and composition methods that Rx exposes.

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Back to the contents page for Reactive Extensions for .NET Introduction

Back to the previous post; Part 3 - Lifetime management – Completing and Unsubscribing

Forward to next post; Part 5 - Combining multiple IObservable streams

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Rx Part 3 – Lifetime management – Completing and Unsubscribing

2010-05-27T19:47:00.001+01:00

So far we have discovered the basics of the Reactive Framework which allows us to create, subscribe and perform some basic aggregations, buffering and time shifting over implementations of IObservable<T>. We have yet to look at how we could unsubscribe from a subscription. If you were to look for an Unsubscribe method in the Rx public API you would not find any. Instead of supplying an Unsubscribe method, Rx will return an IDisposable when ever a subscription is made. This disposable can be thought of as the subscription itself and therefore disposing it will dispose the subscription and effectively unsubscribe. Note that calling Dispose on the result of a Subscribe call will not affect the underlying IObservable<T>, just the instance of the subscription to the IObservable<T>. This then allows us to call Subscribe many times on a single IObservable<T>, allowing subscriptions to come an go with out affecting each other. In this example we initially have two subscriptions, then we dispose of one subscription early which still allows the other to continue to receive publications from the underlying IObservable<T>.

var interval = Observable.Interval(TimeSpan.FromMilliseconds(100));
var firstSubscription =
    interval.Subscribe(value => Console.WriteLine("1st subscription recieved {0}", value));
var secondSubscription =
    interval.Subscribe(value => Console.WriteLine("2nd subscription recieved {0}", value));

Thread.Sleep(500);
firstSubscription.Dispose();
Console.WriteLine("Disposed of 1st subscription");

Console.ReadKey();
/*Outputs:
1st subscription recieved 0
2nd subscription recieved 0
1st subscription recieved 1
2nd subscription recieved 1
1st subscription recieved 2
2nd subscription recieved 2
1st subscription recieved 3
2nd subscription recieved 3
2nd subscription recieved 4
Disposed of 1st subscription
2nd subscription recieved 5
2nd subscription recieved 6
2nd subscription recieved 7

etc....
*/

In the above example, it looks like the values are being produced by the interval Observable by a single OnNext call, however these are independent and work similarly to how a Observable.Create<T> method would work. In this sample we just pause a bit before making our second subscription. Note that the output is different to the above example.

var interval = Observable.Interval(TimeSpan.FromMilliseconds(100));

var firstSubscription =
    interval.Subscribe(value => Console.WriteLine("1st subscription recieved {0}", value));
Thread.Sleep(500);
var secondSubscription =
    interval.Subscribe(value => Console.WriteLine("2nd subscription recieved {0}", value));

Thread.Sleep(500);
firstSubscription.Dispose();
Console.WriteLine("Disposed of 1st subscription");
/*
Ouput:
1st subscription recieved 0
1st subscription recieved 1
1st subscription recieved 2
1st subscription recieved 3
1st subscription recieved 4
2nd subscription recieved 0 
1st subscription recieved 5
2nd subscription recieved 1
1st subscription recieved 6
1st subscription recieved 7
2nd subscription recieved 2
1st subscription recieved 8
2nd subscription recieved 3
Disposed of 1st subscription
2nd subscription recieved 4
2nd subscription recieved 5
2nd subscription recieved 6
etc...

*/

The benefits of using the IDisposable Type instead of creating a new ISubscription/IUnsubscription interface or amending the IObservable<T> interface to have an Unsubscribe method is that you get all of these things for free:

The type already exists
people understand the type
IDisposable has standard usages and patterns
Language support via the Using keyword
Static analysis tools like FxCop can help you with its usage.

OnError and OnCompleted()

Both the OnError and OnCompleted signify the completion of a stream. If your stream publishes a OnError or OnCompleted it will be the last publication and no further calls to OnNext can be performed. In this example we try to publish an OnNext call after an OnCompleted and the OnNext is ignored.

var subject = new Subject<int>();
subject.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed"));
subject.OnCompleted();
subject.OnNext(2);

Of course you could implement your own IObservable<T> that did allow publishing after an OnComplete or an OnError, however it would not follow the precedence of the current Subject types and would be a non standard implementation. I think it would be safe to say that the inconsistent behaviour would cause unpredictable behaviour in the applications that consumed your code.

An interesting thing to consider is that when a stream completes or errors, you should still dispose of you subscription. This can make for messy code, but we will discuss best practices in a later post.

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Back to the contents page for Reactive Extensions for .NET Introduction

Back to the previous post; Part 2 - Static and extension methods

Forward to next post; Part 4 - Flow control

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Rx part 2 – Static and Extension methods

2010-05-23T10:23:00.005+01:00

In the previous post we looked at the core types to the Rx library and how we could get up and running with “Observables”. The key types we discussed were the IObservable<T> interface, IObserver<T>, Subject<T> and its siblings ReplaySubject<T>, BehaviorSubject<T> and AsyncSubject<T>. In this post we are going to explore some of the Static methods on the Observable type and some Extension methods to the IObservable<T> interface.

We have already seen on extension method in the last post which was the Overload to Subscribe which allowed us to pass just an Action<T> to be performed when OnNext was invoked instead of passing an IObserver<T>. The other overloads of Subscribe are also very useful

public static class ObservableExtensions
{
    public static IDisposable Subscribe<TSource>(this IObservable<TSource> source);
    public static IDisposable Subscribe<TSource>(this IObservable<TSource> source, Action<TSource> onNext);
    public static IDisposable Subscribe<TSource>(this IObservable<TSource> source, Action<TSource> onNext, Action<Exception> onError);
    public static IDisposable Subscribe<TSource>(this IObservable<TSource> source, Action<TSource> onNext, Action onCompleted);
    public static IDisposable Subscribe<TSource>(this IObservable<TSource> source, Action<TSource> onNext, Action<Exception> onError, Action onCompleted);
}

Each of these overloads allows you to pass various combinations of delegates that you want executed for each publication event an IObservable<T> could raise. You will use these overloads a lot but there are plenty of other static methods that are very helpful on the static Observable class.

Creation and Generation of Observables

The Observable class has several methods for conveniently creating common types of Observables. The most simple are shown below with an example of how to use it and what you would have to write to emulate the method.

Observable.Empty<T>() This returns IObservable<T> that just publishes an OnCompleted.

var subject = new ReplaySubject<string>();
subject.OnCompleted();
//Or
var empty = Observable.Empty<string>();

Observable.Return<T>(T). This method will return an Observable that publishes the value of T supplied and then publish OnCompleted.

var subject = new ReplaySubject<string>();
subject.OnNext("Value");
subject.OnCompleted();
//Or
var obReturn = Observable.Return("Value");

Observable.Never<T>(). This method will return an IObservable<T> but will not publish any events.

var subject = new Subject<string>();
//Or
var never = Observable.Never<string>();

Observable.Throw<T>(Exception). This method just publishes the exception passed.

var subject = new ReplaySubject<string>();
subject.OnError(new Exception());
//Or
var throws = Observable.Throw<string>(new Exception());

Observable.Create<T>(Func<IObserver<T>,IDisposable>) is a little different to the above creation methods. The method signature itself is a bit nasty but once you get to know him, he is not that bad. Essentially this method allows you to specify a delegate that will be executed anytime a subscription is made. The IObserver<T> that made the subscription will be passed to your delegate so that you can call the OnNext/OnError/OnCompleted methods as you need. Your delegate is a Func that returns an IDisposable. This IDisposable will have it’s Dispose method called when the subscriber disposes from their subscription. We will cover unsubscribing/disposing in a future post.

The big benefit of using the Create method is that your method will be a non blocking call which is the whole point of the Rx framework. In this example we show how we might first return an Observable via standard blocking call, and then we show the correct way to return an Observable without blocking

private IObservable<string> BlockingMethod()
{
  var subject = new ReplaySubject<string>();
  subject.OnNext("a");
  subject.OnNext("b");
  subject.OnCompleted();
  Thread.Sleep(1000);
  return subject;
}
private IObservable<string> NonBlocking()
{
  return Observable.Create<string>(
      observable =>
      {
        observable.OnNext("a");
        observable.OnNext("b");
        observable.OnCompleted();
        Thread.Sleep(1000);
        return Disposable.Create(() => Console.WriteLine("Observer has unsubscribed"));
        //or can return an Action like
        //return () => Console.WriteLine("Observer has unsubscribed");
      });
}

I would imagine that you will use Observable.Create<T> method a lot once you start using Rx. An obvious application of this method is if you needed to perform a slow request to the network (web service or SQL call). There is also a Observable.Create<T>(Func<IObserver<T>,Action>) method that instead of invoking the Dispose method it will just invoke the action when the subscriber disposes their subscription.

Observable.Range(int, int) returns just a range of integers. The first integer is the initial value and the second is the number of values to yield. This example will write the values "10” through to “24” and then “Completed”.

var range = Observable.Range(10, 15);
range.Subscribe(Console.WriteLine, ()=>Console.WriteLine("Completed"));

Observable.Interval(TimeSpan) will publish incremental values from 0 every period that you supply. This examples publishes values every 250milliseconds

var interval = Observable.Interval(TimeSpan.FromMilliseconds(250));
interval.Subscribe(Console.WriteLine);

In this example I use the Observable.Interval method to fake ticking prices

var rnd = new Random();
var lastPrice = 100.0;
var interval = Observable.Interval(TimeSpan.FromMilliseconds(250))
    .Select(i =>
      {
          var variation =
              rnd.NextDouble() - 0.5;
          lastPrice += variation;
          return lastPrice;
      });

interval.Subscribe(Console.WriteLine);

Observable.Start method allows you to turn a long running Func<T> or Action into an Observable. If you use the overload that takes an Action then the returned Observable will be of type IObservable<Unit>. Unit is analogous to void and is a Functional Programming construct. In this case it is just used to publish an acknowledgement that the Action is complete, however this is rather inconsequential as OnCompleted will be published straight after Unit anyway. If the overload you use is a Func<T> then when the Func returns its value it will be published and then OnCompleted. Below is an example of using both overloads.

static void StartAction()
{
    var start = Observable.Start(() =>
    {
        Console.Write("Working away");
        for (int i = 0; i < 10; i++)
        {
            Thread.Sleep(100);
            Console.Write(".");
        }
    });
    start.Subscribe(unit => Console.WriteLine("Unit published"), () => Console.WriteLine("Action completed"));
}
static void StartFunc()
{
    var start = Observable.Start(() =>
    {
        Console.Write("Working away");
        for (int i = 0; i < 10; i++)
        {
            Thread.Sleep(100);
            Console.Write(".");
        }
        return "Published value";
    });
    start.Subscribe(Console.WriteLine, () => Console.WriteLine("Action completed"));
}

ToObservable<T>(this IEnumerable<T>) is an extension method for IEnumerable<T> that will translate any IEnumerable<T> to an IObservable<T>. It is a simple an convenient method for switching from one paradigm to another.

var enumT =  new List<string>();
enumT.Add("a");
enumT.Add("b");
enumT.Add("c");
var fromEnum = enumT.ToObservable();

fromEnum.Subscribe(Console.WriteLine);

ToEnumerable<T>(this IObservable<T>) effectively is the opposite of the above method. It will take an IObservable<T> and yield its results as an IEnumerable. Again useful for switching paradigms.

Observable.Generate provides a more elaborate way to construct Observables than Range. In this example I keep it simple and have the Generate method construct me a sequence starting from 5 and incrementing by 3 while the value is less than 15. I also have the values published as strings to show that the type of the State and Result can be different.

var generated = Observable.Generate(5, i => i < 15, i => i + 3, i => i.ToString());
generated.Subscribe(Console.WriteLine, ()=>Console.WriteLine("Completed"));

Checking for existence

Theses method below all provide some sort of check to confirm the existence of a published value.

Any<T>(this IObservable<T>) is an extension method that will either check that the IObservable will publish any value at all, or if you use the overload that take a Func<T,bool> then it will check if any values satisfy your predicate. The interesting thing about Any is that it is non-blocking and so returns an IObservable<bool>. This will only yield one value.

var range = Observable.Range(10, 15);
range.Any().Subscribe(Console.WriteLine);
range.Any(i => i > 100).Subscribe(Console.WriteLine);
Observable.Empty<int>().Any().Subscribe(Console.WriteLine);

Contains<T>(this IObservable<T>, T) is an extension method like Any but will only accept a value of T not a predicate of T. Like Any the result is an IObservable<bool>.

var range = Observable.Range(10, 15);
range.Contains(15).Subscribe(Console.WriteLine);
Observable.Empty<int>().Contains(5).Subscribe(Console.WriteLine);

Filtering and Aggregating

These next few extension methods provide some sort of filter to the underlying stream. These are very similar to the extension methods available to IEnumerable<T>

First<T>(this IObservable<T>) simply returns the first value of an Observable or throws InvalidOperationException if the stream completes without yielding a value. This is a blocking call that wont return until a value or OnCompleted is published.

var range = Observable.Range(10, 15);
Console.WriteLine(range.First());
//InvalidOperationException("Sequence contains no elements.")
Console.WriteLine(Observable.Empty<int>().First());

There are further extension methods that probably don't warrant further explanation if I assume you are familiar with the existing IEnumerable<T> extension methods. These include

FirstOrDefault
Last
LastOrDefault
Single
Count
Min
Max
Sum
Where
GroupBy

While I think that most of these are self explanatory, what maybe worth reinforcing is that some of the scalar methods (returns bool or T and not IObservable) are blocking calls. This can be very handy for transforming your non blocking Rx code into sequential blocking code.

Aggregate & Scan extension method overloads allow you to create your own aggregations. Both take similar arguments for their overloads. One of Scan’s overload takes a Func<T,T,T> which is the more simple overload. Effectively given an accumulator and the current value you must return the next accumulator. The other overload can take another type for the accumulator and allows you to specify the seed value for accumulator.

Aggregate has the same overloads. The difference is that Scan will OnNext the result of every Accumulation, Aggregate will only OnNext the last value. Therefore Aggregate is like an AsyncSubject<T> version of Scan.

In this example Scan will publish the running sum of the input sequence, and Aggregate will only publish the final total.

public void Aggregate_can_be_used_to_create_your_own_aggregates()
{
  var interval = Observable.Interval(TimeSpan.FromMilliseconds(150))
                          .Take(10);  //Must complete for Aggregate to publish a value.
  var aggregate = interval.Aggregate(0L, (acc, i) => acc + i);
  WriteStreamToConsole(aggregate, "aggregate");
}

public void Scan_allows_custom_rolling_aggregation()
{
  var interval = Observable.Interval(TimeSpan.FromMilliseconds(150));
  var scan = interval.Scan(0L, (acc, i) => acc + i);
  WriteStreamToConsole(scan, "scan");
}

Take<T>(this IObservable<T>, int) will return you the first number of publications specified by the integer value provided. Take(1) is like First() expect that is returns an IObservable so is not blocking. This example yields just the values 10 and 11.

var range = Observable.Range(10, 15);
range.Take(2).Subscribe(Console.WriteLine);

Skip<T>(this IObservable<T>, int) will ignore the first number of publications specified by the integer value provided. So while Take(2) above returned 10 and 11, Skip(2) will ignore 10 and 11 and return the rest of the stream.

var range = Observable.Range(10, 15);
range.Skip(2).Subscribe(Console.WriteLine);

DistinctUntilChanged<T>(this IObservable<T>) formerly HoldUntilChanged, will ignore any value that is the same as the previous value. In this example the values a, b and c are only written to the console once each.

var subject = new Subject<string>();
subject.DistinctUntilChanged().Subscribe(Console.WriteLine);
subject.OnNext("a");
subject.OnNext("a");
subject.OnNext("a");
subject.OnNext("b");
subject.OnNext("b");
subject.OnNext("b");
subject.OnNext("b");
subject.OnNext("c");
subject.OnNext("c");
subject.OnNext("c");

Buffering and time shifting

In some scenarios your application or user may not be able to keep up with the amount of data that a stream is providing. Alternatively, your user or application may just not need the amount of data that is being published. Rx offer some great extension methods for allowing you to control the rate at which values are published to your stream.

Buffer allows you to buffer a range of values and then re-publish them as a list once the buffer is full. You can buffer a specified number of elements or buffer all the values per timespan. Buffer also offer more advanced overloads.

static void Main(string[] args)
{
    var range = Observable.Range(10, 15);
    range.Buffer(4).Subscribe(
        enumerable =>
           {
               Console.WriteLine("--Buffered values");
               enumerable.ToList().ForEach(Console.WriteLine);
           }, () => Console.WriteLine("Completed"));

    var interval = Observable.Interval(TimeSpan.FromMilliseconds(150));
    interval.Buffer(TimeSpan.FromSeconds(1)).Subscribe(
        enumerable =>
        {
            Console.WriteLine("--Buffered values");
            enumerable.ToList().ForEach(Console.WriteLine);
        }, () => Console.WriteLine("Completed"));

    Console.ReadKey();
}

Delay is an extension method overload that will time-shift the entire Observable by the TimeSpan specified, or until the DateTime. This means that if you get a stream with 5 values in the first second and then 10 values in the 3rd second, using Delay with a timespan of 2 seconds will yield results in second 3 and 5. If you use the DateTime overload the values will be yielded with in the first second of the DateTime and then in the 3rd second after the DateTime. This example will delay by 2 seconds and also in one minute.

var inOneMinute = DateTime.Now.AddMinutes(1);
var range = Observable.Range(10, 15);
range.Delay(TimeSpan.FromSeconds(2)).Subscribe(Console.WriteLine);
range.Delay(inOneMinute).Subscribe(Console.WriteLine);

Sample<T>(TimeSpan) will just take a one sample value for every specified TimeSpan and publish that value. Quite simple, quite nice.

var interval = Observable.Interval(TimeSpan.FromMilliseconds(150));
interval.Sample(TimeSpan.FromSeconds(1)).Subscribe(Console.WriteLine);

Throttle is similar to the sample and buffer methods in that it is a time based filter. It is more like sample in that it returns a single value (not an array of values like Buffer). Unlike sample, the window that the values are sampled over is not fixed, it slides every time the underlying stream publishes a value. The throttled stream will only produce a value when the underlying stream does not publish a value for the given period, in which case the last value will be published. When a value is published, the timespan will start and if another value is published the value will only be republished if the timespan has elapsed. Regardless of whether the value is published or not the timespan is reset. This means that if the underlying stream always produces values more frequently than the throttle period then Throttle will never publish any results.

The most obvious usage of this to me is holding back on sending a "Google suggest" request to the server as a user is still typing. We could throttle the keystrokes so that we only send up the search string when the user pauses for given period.

In this example all values are ignored as the publisher is producing values quicker than the throttle allows.

var interval = Observable.Interval(TimeSpan.FromMilliseconds(100));
interval.Throttle(TimeSpan.FromMilliseconds(200))
    .Subscribe(Console.WriteLine);

This example that varies the rate at which it produces value (emulating sporadic keystrokes), we will get some values written to the console where the delay is more than the throttle period.

var interval = Observable.Create<int>(
    o =>
     {
         for (int i = 0; i < 100; i++)
         {
             o.OnNext(i);
             Thread.Sleep(i++ % 10 < 5 ? 100 : 300);
         }
         return () => { };
     });
interval.Throttle(TimeSpan.FromMilliseconds(200))
    .Subscribe(Console.WriteLine);

In later posts, Buffer is covered in more detail and we expand on more complex time based operators. Also when you introduce concurrency (which these time based operators do) you need to start to understand the Scheduling semantics of Rx which are also covered in later posts.

Most of these methods are simple and pretty easy to understand. I found my problem with these was the vast number of them. I also appears the the API is less prone to changes in recent releases. If you want to get to know these methods a bit better check out these links, but most importantly – Use them. Have a play in LinqPad or just some console app.

Useful links:

Rx Home
Exploring the Major Interfaces in Rx – MSDN
IObservable<T> interface - MSDN
IObserver<T> interface - MSDN
Observer Design pattern - MSDN
ObservableExtensions class - MSDN
Observable class - MSDN
Observer class - MSDN
Qservable class - MSDN
Action<T> Delegate - MSDN
Func<T1, T2, TResult> - MSDN
The Rx Wiki site 101 Samples
Enumerate This
LinqPad

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

Back to the contents page for Reactive Extensions for .NET Introduction

Back to the previous post; Part 1 - Introduction to Rx

Forward to next post; Part 3 - Lifetime management – Completing and Unsubscribing

Technorati Tags: Reactive-Extensions,Rx,Tutorial

Introduction to Rx Part 1 - Key types

2010-05-18T23:06:00.001+01:00

Microsoft has released a new library for building “reactive” applications. It’s full name is Reactive Extensions for .NET but is generally referred to as just “Rx”. Essentially Rx is built upon the foundations of the Observer pattern. .NET already exposes some other ways to implement the Observer pattern such as Multicast delegates or Events. Multicast delegates (which Events are) however can be cumbersome to use, have a nasty interface and are difficult to compose and can not be queried. Rx looks to solve these problems.

Here I will introduce you to the building blocks and some basic types that make up Rx.

IObservable<T>

IObservable<T> is one of the 2 core interfaces for working with Rx. It is a simple interface with just a Subscribe method. Microsoft are so confident that this interface will be of use to you it has been included in the BCL as of version 4.0 of .NET. You should be able to think of anything that implements IObservable<T> as a Stream of T objects. So if a method returned an IObservable<Price> I could think of it as a stream of Prices.

IObserver<T>

IObserver<T> is the other one of the 2 core interfaces for working with Rx. It too has made it into the BCL as of .NET 4.0. Don’t worry if you are not on .NET 4.0 yet as the Rx team have included these 2 interfaces in a separate assembly for .NET 3.5 users. IObserver<T> is meant to be the “functional dual of IEnumerable<T>”. If you want to know what that last statement meant then enjoy the hours of videos on Channel9 where they discuss the mathematical purity of the types. For everyone else it means that where an IEnumerable<T> can effectively yield 3 things (the next value, an exception or the end of the sequence), so too can IObservable<T> via IObserver<T>’s 3 methods OnNext(T), OnError(Exception) and OnCompleted().

Interestingly, while you will be exposed to the IObservable<T> interface a lot if you work with Rx, I find I don't often need to concern myself with IObserver<T>. Another interesting thing I have found with Rx is that I never actually implement these interfaces myself, Rx provides all of the implementations I need out of the box. Lets have a look at the simple ones.

Subject<T>

If you were to create your own implementation of IObservable<T> you may find that you need to expose method to publish items to the subscribers, throw errors and notify when the stream is complete. Hmmm they all sound like the methods on the IObserver<T> interface. While it may seem odd to have one type implementing both interfaces, it does make life easy. This is what subjects can do for you. Subject<T> is the most basic of the subjects. Effectively you can expose your Subject<T> behind a method that returns IObservable<T> but internally you can use the OnNext, OnError and OnCompleted methods to control the stream.

In this (awfully basic) example, I create a subject, subscribe to that subject and then publish to the stream.

using System;
using System.Collections.Generic;

namespace RxConsole
{
    class Program
    {
        static void Main(string[] args)
        {
            var subject = new Subject<string>();

            WriteStreamToConsole(subject);

            subject.OnNext("a");
            subject.OnNext("b");
            subject.OnNext("c");
            Console.ReadKey();
        }

        private static void WriteStreamToConsole(IObservable<string> stream)
        {
            stream.Subscribe(Console.WriteLine);
        }
    }
}

Note that the WriteStreamToConsole method takes an IObservable<string> as it only wants access to the subscribe method. Hang on, doesn’t the Subscribe method need an IObserver<string>? Surely Console.WriteLine does not match that interface. Well not it doesn’t but the Rx team supply me with an Extension Method to IObservable<T> that just takes an Action<T>. The action will be executed every time an item is published. There are other overloads to the Subscribe extension method that allows you to pass combinations of delegates to be invoke for OnNext, OnCompleted and OnError. This effectively means I don't need to implement IObserver<T>. Cool.

As you can see, Subject<T> could be quite useful for getting started in Rx programming. Subject<T> is a basic implementation however. There are 3 siblings to Subject<T> that offer subtly different implementations which can drastically change the way your program runs.

ReplaySubject<T>

ReplaySubject<T> will remember all publications to it so that any subscriptions that happen after publications have been made, will still get all of the publications. Consider this example where we have moved our first publication to occur before our subscription

static void Main(string[] args)
{
    var subject = new Subject<string>();

    subject.OnNext("a");
    WriteStreamToConsole(subject);
    
    subject.OnNext("b");
    subject.OnNext("c");
    Console.ReadKey();
}

The result of this would be that “b” and “c” would be written to the console, but “a” ignored. If we were to make the minor change to make subject a ReplaySubject<T> we would see all publications again.

static void Main(string[] args)
{
    var subject = new ReplaySubject<string>();

    subject.OnNext("a");
    WriteStreamToConsole(subject);
    
    subject.OnNext("b");
    subject.OnNext("c");
    Console.ReadKey();
}

This can be very handy for eliminating race conditions.

BehaviorSubject<T>

BehaviorSubject<T> is similar to ReplaySubject<T> except it only remembers the last publication. BehaviorSubject<T> also requires you to provide it a default value of T. This means that all subscribers will receive a value immediately (unless it is already completed).

In this example the value “a” is written to the console.

static void Main(string[] args)
{
    var subject = new BehaviorSubject<string>("a");
    WriteStreamToConsole(subject);
    Console.ReadKey();
}

In this example the value “b” is written to the console, but not “a”.

static void Main(string[] args)
{
    var subject = new BehaviorSubject<string>("a");
    subject.OnNext("b");
    WriteStreamToConsole(subject);
    Console.ReadKey();
}

In this example the values “b”, “c” & “d” are all written to the console, but again not “a”

static void Main(string[] args)
{
    var subject = new BehaviorSubject<string>("a");

    subject.OnNext("b");
    WriteStreamToConsole(subject);
    subject.OnNext("c");
    subject.OnNext("d");
    Console.ReadKey();
}

Finally in this example, no values will be published as the stream has completed. Nothing is written to the console.

static void Main(string[] args)
{
    var subject = new BehaviorSubject<string>("a");

    subject.OnNext("b");
    subject.OnNext("c");
    subject.OnCompleted();
    WriteStreamToConsole(subject);
    
    Console.ReadKey();
}

AsyncSubject<T>

AsyncSubject<T> is similar to the Replay and Behavior subjects, however it will only store the last value, and only publish it when the stream is completed.

In this example no values will be published so no values will be written to the console.

static void Main(string[] args)
{
    var subject = new AsyncSubject<string>();

    subject.OnNext("a");
    WriteStreamToConsole(subject);
    subject.OnNext("b");
    subject.OnNext("c");
    Console.ReadKey();
}

In this example we invoke the OnCompleted method and the value “c” is published and therefore written to the console.

static void Main(string[] args)
{
    var subject = new AsyncSubject<string>();

    subject.OnNext("a");
    WriteStreamToConsole(subject);
    subject.OnNext("b");
    subject.OnNext("c");
    subject.OnCompleted();
    Console.ReadKey();
}

So that is the very basics of Rx. With only that under you belt it may be hard to understand why Rx is a topic of interest. To follow on from this post I will discuss further fundamentals to Rx

Extension methods
Scheduling / Multithreading
LINQ syntax

Once we have covered these it should allow you to really get Rx working for you to produce some tasty Reactive applications. Hopefully after we have covered these background topics we can knock up some Samples where Rx can really help you in your day to day coding.

The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.

MergedDictionaries performance problems in WPF

2010-05-12T19:14:00.001+01:00

I don’t normally like to blatantly plagiarise other people’s comments, but this seems to be a little know bug that sounds like it should be shared.

A colleague of mine emailed our internal tech list the following email

I strongly urge everyone working with WPF to use this or at least benchmark it in your own applications if you use ResourceDictionaries.MergedDictionaries. I consider this to be a huge problem in WPF. I’m not sure if it exists in Silverlight, but I would assume it does.

I was just debugging a very long render delay in some WPF code and I came across this little tidbit:

http://www.wpftutorial.net/MergedDictionaryPerformance.html

The quote of interest is: “Each time a control references a ResourceDictionary XAML creates a new instance of it. So if you have a custom control library with 30 controls in it and each control references a common dictionary you create 30 identical resource dictionaries!”

Normally that isn’t a huge problem, but when you consider the way that I personally (and have suggested to others) that they organize their resources in Prism projects it gets to be a **serious** problem. For example, let’s say we have this project structure:
/MyProject.Resources
       /Resources
                -Buttons.xaml
                -DataGrid.xaml
                -Global.xaml
                -Brushes.xaml
                -WindowChrome.xaml
                -Icons.xaml
 
/MyProject.Module1
      /Resources
                -Module1Resources.xaml  (References all Dictionaries in /MyProject.Resources/Resources/*)
      /Views
                -View1.xaml
                -View2.xaml
      
/MyProject.Module2
      /Resources
                -Module2Resources.xaml   (References all Dictionaries in /MyProject.Resources/Resources/*)
      /Views
                -View1.xaml
                -View2.xaml
      
/MyProject.Shell
      /Resources
                -ShellResources.xaml   
      /Views
                -MainShell.xaml
If in your views you reference the module-level ResourceDictionary (which helps for maintainability and modularity) then every time you create an instance of View1.xaml for example, you would have to parse all the ResourceDictionaries in /MyProject.Resources/Resources/* every time. This isn’t really a memory concern but it is a huge performance concern. There can potentially be thousands of lines of XAML code to parse and the time really does add up.

I recently switched all of the MergedDictionary references:
<ResourceDictionary>
    <ResourceDictionary.MergedDictionaries>
        <ResourceDictionary Source=”/SomeDictionary.xaml” />
    </ResourceDictionary.MergedDictionaries>
</ResourceDictionary>
To use the attached SharedResourceDictionary which shadows the Source property and keeps a global cache of all ResourceDictionaries parsed:
<ResourceDictionary>
    <ResourceDictionary.MergedDictionaries>
        <SharedResourceDictionary Source=”/SomeDictionary.xaml” />
    </ResourceDictionary.MergedDictionaries>
</ResourceDictionary>
And I saw a performance increase of almost two orders of magnitude … From almost 6000ms to 200ms. I’ve attached this code; I used the basic sample implementation in the link above so this is considered public information for client purposes.

Cheers,

Charlie

Thanks to Charlie Robbins (Lab49) for expanding on Christian’s blog post and for letting me re-print your email.

Squeezing out performance from Charting

2010-02-13T01:25:00.001Z

After my review of the charting products available currently I decided to go with the Slverlight/WPF Data Visualization project (from the WPF Toolkit). I did end up coming up with a trick to squeeze some performance out of the charts. Quite simply that charts don’t handle a lot of data very well. When you have rich data templates, animation and more than several hundred data points, performance is pretty poor. I decided that the first thing to compromise on would be animation. It is nice but I would rather speed. By turning off the animation in the charting I get a little speed up but there still is a simple truth that must be considered. If my graph is only 500px wide then why try to render more than 500 data points? This is my first step to gaining some performance; filter out data by sampling so we never try to get the chart to render more data than it ever could.

I can achieve this by creating a custom CollectionViewSource. The new sub class is simple;

it has a dependency property of MaxItemCount
on any change to the MaxItemCount or the Source we sample the data set, and save the values we want to display into a set
we subscribe to the Filter event and only accept an item if it is included in our sample set

public class CollectionSizeFilter : CollectionViewSource
{
    int _count;
    ICollectionView _defaultView;
    HashSet<object> _toKeep;

    public CollectionSizeFilter()
    {
        Filter += CollectionSizeFilter_Filter;
    }

    protected virtual void CollectionSizeFilter_Filter(object sender, FilterEventArgs e)
    {
        e.Accepted = _toKeep == null || _toKeep.Contains(e.Item);
    }

    protected override void OnSourceChanged(object oldSource, object newSource)
    {
        base.OnSourceChanged(oldSource, newSource);
        _defaultView = GetDefaultView(newSource);
        _count = Count(_defaultView.SourceCollection);

        LoadHashset();
    }

    public double MaxItemCount
    {
        get { return (double)GetValue(MaxItemCountProperty); }
        set { SetValue(MaxItemCountProperty, value); }
    }
    public static readonly DependencyProperty MaxItemCountProperty = DependencyProperty.Register("MaxItemCount", typeof(double), typeof(CollectionSizeFilter), new UIPropertyMetadata(1d, MaxItemCountProperty_Changed));

    private static void MaxItemCountProperty_Changed(DependencyObject sender, DependencyPropertyChangedEventArgs e)
    {
        var self = (CollectionSizeFilter)sender;
        self.LoadHashset();
    }

    private void LoadHashset()
    {
        if (_count <= MaxItemCount)
        {
            _toKeep = null;
        }
        else
        {
            _toKeep = new HashSet<object>();
            var gap = MaxItemCount - 1;
            var spacing = _count / gap;
            double nextIndex = 0d;
            int i = 0;
            foreach (var item in _defaultView.SourceCollection)
            {
                if (i >= nextIndex)
                {
                    _toKeep.Add(item);
                    nextIndex += spacing;
                }
                i++;
            }
        }
        if (View != null)
            View.Refresh();
    }

    private static int Count(IEnumerable source)
    {
        if (source == null)
        {
            return 0;
        }
        var is2 = source as ICollection;
        if (is2 != null)
        {
            return is2.Count;
        }
        int num = 0;
        IEnumerator enumerator = source.GetEnumerator();
        {
            while (enumerator.MoveNext())
            {
                num++;
            }
        }
        return num;
    }
}

To use the new “control” we just treat it like a normal CollectionViewSource but we specifiy the MaxItemCount by binding it to the width of the Chart like this.

<Controls:CollectionSizeFilter x:Key="FilteredData" 
    Source="{Binding MyData}" 
    MaxItemCount="{Binding ElementName=chart1, Path=ActualWidth}"/>

This gave some performance improvements but I thought why not follow this concept down the path a little bit more. For most data that I want to display I am mainly interested in the trend not the minutia. So why do I try to show a data point on every pixel? I could just sample the data; for example if I have a data set of 1500 data points and a graph that is 500px wide, I get some performance gains by reducing the rendered data set to 500 data points, but why not just show one data point every say 10px? If this is acceptable for your data then you can reduce your rendered data set from 1500 down to 50 (30 times less data to render). To do this is even more simple than the code above. We just need to create and implementation of IValueConverter to do some division, a DivisionConverter.

public sealed class DivisionConverter : IValueConverter
{
    #region IValueConverter Members
    public object Convert(object value, Type targetType, object parameter, CultureInfo culture)
    {
        double numerator = ConvertToDouble(value, culture);
        double denominator = ConvertToDouble(parameter, culture);
        return numerator / denominator;
    }

    public object ConvertBack(object value, Type targetType, object parameter, CultureInfo culture)
    {
        throw new NotImplementedException();
    }
    #endregion

    private static double ConvertToDouble(object value, IFormatProvider culture)
    {
        var result = default(double);
        try
        {
            var source = value as IConvertible;
            if (source != null)
                result = source.ToDouble(culture);
        }
        catch
        {
        }
        return result;
    }
}

So now we can just extend our previous XAML to include the new converter and apply our sample size of 10.

<Controls:DivisionConverter x:Key="divisionConverter" />
<Controls:CollectionSizeFilter x:Key="FilteredData" 
    Source="{Binding MyData}"
    MaxItemCount="{Binding ElementName=chart1, Path=ActualWidth, 
        Converter={StaticResource divisionConverter}, 
        ConverterParameter=10}"/>

Putting it all together I end up with code like below. I prefer to keep my concerns near each other, so here I have defined the filters in the LineSeries Resources instead of where some may put them which is the top of the file. I would however define the key to the DivisionConverter at the top of the file or even in the App.xaml as I will probably use it in many places. Note how I use RelativeSource Binding to find the width of the parent chart.

<chartingToolkit:Chart Name="chart1">
    <chartingToolkit:LineSeries 
        Title="{Binding Title}"
        DependentValuePath="Value"
        IndependentValuePath="Date">
        <chartingToolkit:LineSeries.Resources>
            <Controls:CollectionSizeFilter 
                x:Key="FilteredBalances" 
                Source="{Binding Balances}"
                MaxItemCount="{Binding 
                        RelativeSource={RelativeSource FindAncestor, 
                        AncestorType={x:Type chartingToolkit:Chart}}, 
                        Path=ActualWidth,
                        Converter={StaticResource DivisionConverter}, 
                        ConverterParameter=10}"/>
        </chartingToolkit:LineSeries.Resources>
        <chartingToolkit:LineSeries.ItemsSource >
            <Binding Source="{StaticResource FilteredBalances}"/>
        </chartingToolkit:LineSeries.ItemsSource>
    </chartingToolkit:LineSeries>
</chartingToolkit:Chart>

You may find your millage varies with the sampling size. You may be only comfortable with small values like 3-5 or you may be more aggressive with values around 30. It is your data and you will know what is best for you. The real beauty of this is that problem of performance is a presentation problem. With some simple controls we are able to tame the problem in the presentation layer without having to compromise the purity of our ViewModels (like setting max size values that get sent to databases). Here if we resize the chart, we already have all the data so we just render more data points. Also the solution is very generic, there are no dependencies on the WPF Data Visualization assemblies so you may find other uses for them.

My WPF Charting Comparisons

2010-01-18T07:09:00.002Z

I have recently been looking for some graphing/charting functionality for a home project I am working on. My requirements are fairly simple:

handle data quantities in the region of thousands and tens of thousands of rows/items
be able to display line charts with or without data points (there will be so many data points that they can become noise)
be able to display multiple sets of data to be able to compare data
free or cheap
xcopy install

Now as the Charting products I wanted to compare were all going to be in WPF I assumed that these requirements were just a given but apparently not, so let me specify them as well

be able to bind the data from my own view model (i.e. I don’t want to have controls littering my View Model)
have the graph update as the data changes

Now to see the list of contenders:

WPF Toolkit Charting (aka System.Windows.Controls.DataVisualization.Toolkit)
AmCharts
VisiFire
DynamicDataDisplay (aka D3)

So for the really quick review of each

WPF Toolkit Charting

This is the CodePlex project from some of the lads at Microsoft. This is presumable of a lesser quality than the rest of the Toolkit as the Charting component is in preview. The WPF Toolkit allows for great looking charts by utilising the power of WPF Styles. It is one of those balancing acts that must be difficult to make when designing software; extensibility vs. simplicity. The WPF Toolkit leans more towards the extensible option. Extending the charts to look the way you want can be done but many will find it fiddly and frustrating, but once done can be very rewarding and the Graphs can look amazing. The WPF Toolkit also utilises the power of WPF binding by allowing me to bind to my ViewModel. So it looks like a good start, however, the clear and painful problem with the WPF Toolkit is performance. When loading up even hundreds of rows/items the performance is fairly poor. When I tried to throw just over a thousand items at a Line Series the performance was completely unacceptable. One other problem I have is that I get intermittent lock ups. When updating the data, the charting code will run off into a loop and not come out of it, freezing the UI. Hmmm another cross.

Positives:

Extensibility allows for beautiful graphs
Charts bind to ViewModel
Free

Negatives:

Woeful performance
Random lock ups.

AmCharts

AmCharts appears to be a Charting solution aimed at the financial industry. The chart control that I thought would best fit my needs was the Stock chart. This chart had a great feature that allowed zooming on the X-axis by providing a range slider. Performance was great when I threw ~1500 items at the control. An odd problem I had was the graph would only appear once I resized my window. I think this has to do with binding to a ViewModel as the Demo does not have this problem but it also directly interacts with the control from the Code behind. I want to avoid “messing with controls” from my ViewModel. A more real problem I have is that while the performance is great, the binding seems to be a once off event. Changes to the values in my collection are not reflected by any change to the chart.

Positives

Good performance
Charts bind to ViewModel
Zoom functionality
Good samples
Smallest DLL size (223KB)

Negatives

One time data binding
Odd problem with Chart not rendering until i resized the window.

Visifire

VisiFire charts looked to be a great option. They were very easy to get up and running, had some good samples like the AmCharts. My first play with the Visifire charts provided me with a good looking chart. My problems came when I went to bind the Charts to my ViewModel…Visifire does not support data binding! I’m not even sure why someone would write a WPF control that does not support data binding. I wasted plenty off time writing some adapters so that I could get data binding working. Data binding is in the wish list for version 3 (how it didn't make it into the wish list for the 1st version I don’t know). Performance of Visfire was pretty good (not spectacular) and sat in between AmCharts and the hopelessly slow WPF Toolkit.

Positives

Easy to get up and running
Pretty good looking default charts
Moderate performance

Negatives

You cant bind a data series to a collection!

Dynamic Data Display

The Dynamic Data Display (aka D3) is another Microsoft project on Codeplex from a Microsoft research team in Russia. D3 authors claim outstanding performance even with massive amounts of data. Sounds like a sure fire winner! The control library also supports different types of charts to the other libraries like Maps and Isolines ( I have no idea what an isoline is). The samples show some good stuff with smooth moving animated graphs with dynamic data points. The big fail on the project is again, no data binding. All manipulation of the charts needs to be done in C# code and needs to be very imperative. There are some guys, however, who have made posts creating an extension to the controls to support data-binding. Either way, while this looked to be a good set of controls, the authors don’t appear to have followed the Pit-of-Success principle. I would go in to details, but the fact it took me hours of reading forums, looking at samples and coding to just get my Model showing on the screen. When it did get on to the screen it was fast, but didn’t update when the underlying data changed. This is a very immature set of controls but may have a bright future if the team can get some fundamentals right.

Positives

High performance
Easy to scroll and zoom data

Negatives

Hardest set of controls to work with. Everything has to be done in code. Authors seem to miss the point of WPF entirely. Presentation and logic feel very much couple together.
After all my mucking around the chart didn’t update with my changes to the data.

In summary, I am pretty disappointed with the state of all of these charting controls. What I did manage to get working to a satisfactory state was the WPF Toolkit. As the only real problem I had with the WPF Toolkit charting controls was their performance; I decided that an easy way to get some better performance out of the control was to only show as many data points as there were available pixels. If I only have 400 pixels to show my data it becomes a bit silly to try and get the graph to render 1400 data points. I created a Custom Control that extends CollectionViewSource by having a MaxItemCount property that can be set to effectively filter the amount of data the CollectionViewSource reveals to the Charting controls. The performance was better but I was able to further tweak the performance by adding a DivisionConverter to further reduce the collection size by the parameter specified (10 in my case). This means I only show a data point for every 10 pixels wide the chart is. This ended up being a great compromise….except for the random lock ups. If I play on the Chart for long enough changing the data to update the chart, eventually the program just falls in to a loop. If I can solve this bug I may have a winner on my hands. Ed:—Playing around more I may have got rid of this problem. Still pops up sometimes straight after a build, but a restart fixes it. This may be to do with my build of Win7 (pre release that I am still running). This throws the WPF Toolkit +the 2 tiny bits of filter code clearly into the lead as it can be made to look great and handle tens of thousands of rows.

If any one is interested in the code I used to test/play with each of these libraries you can find a zip of the VS2008 solution here. To see any of the spikes, just set it as the start up project and run or Right click on the project and “Debug”—> “Start new instance”. Only the MyDomain project wont run as that is the Class library that has the small part of the domain to test the charts with.

ChartingPlaygournd.zip – Source code for my tests.

Logging in modern .NET applications

2009-09-23T00:05:00.001+01:00

Background - A common evolution to logging

Recently we needed to implement a decent logging solution for a new application. The application (like so many) had evolved from a proof of concept (P.O.C) code base. 5 weeks after the cut over from P.O.C to code that was intended to ship, we had a plethora of Debug.WriteLine, Console.WriteLine and Trace.WriteLines sprinkled through the code base. Most of the *.WriteLine code was to help debug

code that implemented 3rd party code
code that had no tests (booo!) and F5 testing seemed to be "good enough"

After a couple of preliminary builds were sent to the client for initial evaluation, several un-handled exceptions had crept through. To be fair to the team, most of these were related to some undocumented features of the 3rd party system and few were my fault where maybe I was pushing WPF a bit to far for the clients machines. The result of these exceptions were a custom popup window on the WPF application, stating that a fatal exception had occurred and would shut down now. We would also email the exception as a quick way to notify the dev team of any errors.
At this point in time the team had decided we had several real problems here:

We had various "debugging" systems (Console, debug and trace)
We had hard coded our implementations of "logging". If logging a specific piece of functionality is a requirement we really should test that we are logging the correct thing!
As the WPF guy I am interested in any WPF binding errors (in the VS output window), but the service/agent guy were flooding the console screen with their pub/sub debug messages. How could I filter out their messages, and how could they filter out mine?
We were catching the un-handled exception at the dispatcher, not at the point it was thrown. This mean we had a huge stack trace (noise) pointing back to the problem and we also were missing some useful data to debug the exception such as the inputs to the method that threw the exception.

Setting the requirements

From the problems we have above the team decided that our requirements for the logging system were:

Testable
Provide different levels(severity) of logging. Trace, Debug and Exception.
Filter on logging level
Log to various targets. Email, File and Console.
Provide different areas or categories to log as. e.g. Presentation, Agent, Services...
Filter on logging Category (so I don't see the agent debug messages and agent developers don't see my presentation guff)

The requirements for the implementation of the logging system are:

Log un-handled exceptions so we can analyse then when they occur on live
Log the inputs to the method that threw the exception so we can identify it it is data related
Replace console/debug/trace calls with new logger.

Journey to a solution

Logging System

First thing I will tackle is the testability. I will create my own interface called ILogger that will do my logging for me

public interface ILogger
{
  void Write(string message);
}

Well that was easy! Next, Provide different levels of logging. I am going to do that by replacing my Write method with level specific methods:

public interface ILogger
{
    void Trace(string message);
    void Debug(string message);
    void Error(string message);
}

To now add Categories. Here I will add a category parameter to each of the methods. I would advise against using enums for your categories, especially if your logger becomes a shared API. Enums can not be extended and really tie you down.

public interface ILogger
{
    void Trace(string category, string message);
    void Debug(string category, string message);
    void Error(string category, string message);
}

Hmmm. It now looks like the write we are making a mess for the poor developer who has to implement this interface. I think I may have been on the right track with the write method in the first place. I am going to change that back a Write method that takes 3 parameters. I will offer extension methods to give users the ability to make calls like the one above (a little bit more work for me but less work for everyone else). See here for more info on extending interfaces with Extension methods.

public interface ILogger
{
    void Write(LogLevel level, string category, string message);
}

Some may be thinking that "Hang on. Strings are bad! Surely a category Enum has to be better?". To answer this I would recommend using an internal class like this

class Category
{
    public const string Agent = "Agent";
    public const string Presentation = "Presentation";
    public const string Agent = "Service";
}

If you want to expose categories at your public API, create a public version of this class that consumers can use an even inherit from to add their own categories too. If you are thinking "shouldnt they be public static readonly?", I will cover that later.
Checking off aginst our requirements list we can cross somethings off

~~Testable~~ (Interfaces make things testable)
~~Provide different levels(severity) of logging. Trace, Debug and Exception.~~
Filter on logging level
Log to various targets. Email, File and Console.
~~Provide different areas or categories to log as. eg Presentation, Agent, Services...~~
Filter on logging Category (so I dont see the agent debug messages and agent developers don't see my presentation guff)

So that is a good start. Now we need to add filtering and targeting various outputs. Well luckily there are plenty of 3rd party logging tools out there that do all of this for us. As our project is using Enterprise Library already we will just use their logging application block. Look to the example at the end of the post for complete example.
Great! Thanks to the 3rd party logging system we have ticked off all of our system requirements, now for our implementation requirements.

Implementation of Logging

Now that we have an interface to code against, let us now look at how we would use it.
Ed- This is completely back to front. You should normally look at how you would use it first and then create the implementation. TDD is a great methodology for this. We are approaching it back-to-front in this post because I think it is easier to consume for the reader.
So the main pain point we have is logging exceptions that occur at the agents (the classes that get data from services and map them to client side entities). This is due to poor 3rd party documentation, Live being slightly different to Dev and QA and integration testing being harder to perform than unit testing.
Prior to our new Logging system some agent code might look like this:

public void AmendPortfolio(PortfolioAmmendment portfolio)
{
    Console.WriteLine("AmendPortfolio {0}", portfolio.Id);
    var data = portfolio.MapToServiceData();    //Mapper extension method.
    _service.AmendBasket(_session, data, true);    //Call to 3rd party system
    Console.WriteLine("AmendPortfolio complete");
}

Now if we swapped out the console calls with out logger and then put in some exception logging it may look like this

public void AmendPortfolio(PortfolioAmmendment portfolio)
{
    _logger.Debug(Category.Agent, "AmendPortfolio {0}", portfolio.Id);
    try
    {
        var data = portfolio.MapToServiceData();    //Mapper extension method.
        _service.AmendBasket(_session, data, true);    //Call to 3rd party system
    }
    catch(Exception ex)
    {
        _logger.Error(Category.Agent, ex.ToString);
        throw;
    }
    _logger.Debug(Category.Agent, "AmendPortfolio complete");
}

Oh dear. We now have more code that does logging than code that does real work. While we have satisfied our requirements, we have doubled our work load. Not good. Back to the drawing board.

AOP

Some will have heard of Aspect Orientated Programming. It seemed like 5 years ago it was going to change everything. Well it mainly just changed logging. AOP is a style of programming that allows code to be injected at a given interception point. In our case the interception point would be the start of our agent method, the end of the agent method and when the method throws and exception (which is really the same as the end of the method). As far as I know there are 2 popular ways to achieve this

at run time using a tool like Castle Windsor or Microsoft Unity
at compile time using a tool like PostSharp

I have had some experience with PostSharp as I used it prior to Unity having the ability to add interceptors. So for our solution we went with PostSharp. I believe to switch between any of these options would not be a huge amount of work.
First a quick introduction on how I have previously done AOP logging with PostSharp. I would create an Attribute that I would apply to classes or methods that I wanted to be logged. The attribute would satisfy the requirements of PostSharp so that code would be injected at compile time to do my logging.
code like this

[Logged]
public void AmendPortfolio(PortfolioAmmendment portfolio)
{
    var data = portfolio.MapToServiceData();    //Mapper extension method.
    _service.AmendBasket(_session, data, true);    //Call to 3rd party system
}

which at compile time would alter the IL to represent something more like this:

public void AmendPortfolio(PortfolioAmmendment portfolio)
{
    Logger.Debug("AmendPortfolio({0})", portfolio.Id);
    try
    {
        var data = portfolio.MapToServiceData();    //Mapper extension method.
        _service.AmendBasket(_session, data, true);    //Call to 3rd party system
    }
    catch(Exception ex)
    {
        Logger.Error(ex.ToString);
        throw;
    }
    Logger.Debug("AmendPortfolio complete");
}

Well that looks perfect doesn't it? Not really. We dont have a category specified and we have a hard coded reference the static class Logger from Enterprise Library Logging. It no longer points to our _logger member variable which was of type ILogger. This makes our testing harder to do. If testing your logger is not really something you care about (which is fine) then this AOP solution might be for you. If you do want to be more specific about logging then we need to find a way of getting hold of the instance of ILogger. As PostSharp is a compile time AOP framework, it is a touch harder to integrate than if we used Unity or Windsor. The main problem is how do we get a handle on the Logger? A solution we came up with was to create an ILogged interface

public interface ILogged
{
    ILogger Logger { get; }
}

By doing this we expose the Logger so we can use it in the Aspect/Attribute. Now if we look at out method that we are trying to log in the greater context of the class it resides in we can see what the implementation may now look like.

[Logged]
public class PortfolioAgent : IPortfolioAgent, ILogged
{
    private readonly ILogger _logger;
    private readonly ISomeService _service;
    public PortfolioAgent(ILogger logger, ISomeService _service)
    {
        _logger = logger;
        _service = service;
    }
    public void AmendPortfolio(PortfolioAmmendment portfolio)
    {
        var data = portfolio.MapToServiceData();    //Mapper extension method.
        _service.AmendBasket(_session, data, true);    //Call to 3rd party system
    }
}

That looks kinda cool to me. One thing my colleagues made a note of is that they would prefer if in this scenario we used property injection for the logger. This is fine with me as long as it is easy to use and we can still test it. The ILogged interface does not preclude using property injection, it is just not my preference. Another thing to note is the lack of Category. Easy fix there is to add a property to our LoggedAttribute of string Category.

[Logged(Category=Category.Agent)]
public class PortfolioAgent : IPortfolioAgent, ILogged
{
...
}

If you remember earlier that I mentioned public const vs public static readonly. This is the reason why I choose to use const types so that they could be used in the attribute.

I am pretty happy with this now. I think we tick off all of our requirements and have not added a lot of complexity to our code. The one last thing that bugs me is that I have to use the LoggedAttribute and the ILogged interface as a couple. If I forget to use one without the other, I either will get no logging, a nasty runtime exception or if I code a aspect correctly I can get a compile time error (the most desirable). At first I coded the attribute to do the latter (compile time error), but then realised that all of my agents were in one project and all used the same Category. To make life a little bit easier I moved the attribute to the AssemblyInfo and had the Aspect apply itself automatically to any class the implemented ILogged. This maybe a step too far towards black magic for some teams, so do what fits best.

Have a look at the sample code here. Both styles are shown in the example so take your pick of which suits you best.

Success, Motivation and Management

2009-09-16T15:49:00.001+01:00

I just watched an interesting presentation from Dan Pink on The surprising science of motivation. Here he discusses the conventional ways to get productivity out of employees via carrot and stick mentality. Watch the video first (18min) so I don't take the wind from his sails.

What I found interesting especially on refection of my earlier post Projects – Measuring success and providing boundaries, was how he related the 20th century management style to prescriptive roles. Prescriptive roles are roles where you can give very clear guidance on how to perform a task and boundaries of the role. Boundaries normally define measurable tasks with reward/punishment (carrot/stick) attached to the tasks. These can anything from simple things such as

If you work late the company will pay for pizza delivery
if the project comes in on time you get a bonus
if you meet your KPIs you get all of your bonus. Pessimistically viewed as: miss any of your KPIs and we will dock your pay.

However the interesting thing about prescriptive roles is that in the 21st century the game has changed. Any task that can be completed without a level of creativity, abstract thought or reasoning generally can be done:

cheaper by outsourced labour
faster by automation (mechanical or computer)

This affects the software industry massively. Outsourcing burst on to the scene at the turn of the century and appeared to be the Holy Grail for accountants across western civilisation. This also scared the hell out of any "computer guy" as how was he going to make the payments on his new sports car? Outsourcing was not all that is was cracked up to be with stories of low quality product and communication failures. Outsourcing seems to be making a small come back and I think that we will see this see-saw rock a little bit more before outsourcing becomes part of our day to day life. See the 4 hour working week for some great ideas on outsourcing your life.

Dan Pink discusses that the 20th Century style of carrot/stick management worked well with prescriptive roles. But we are in the 21st century now and I would like to think that any one reading this is not performing a prescriptive role. I would even argue that our role is to eliminate or automate what we can. Normally these things that can be eliminated or automated are prescriptive processes. Roles that add real value to any company do require creativity, problem solving, communication skills, building relationships etc. These things cannot (yet) be automated.

So moving from the production line 20th century to the service orientated 21st century we are seeing a shift from the role of management being one based around compliance (carrot/stick) to self managing, autonomous teams/employees. This is in line with what agile and lean concepts are trying to achieve. Creating a culture where

autonomy
mastery
purpose

are values that are held dear, creates an amazing shift positivity of the team. Instead prescribing exactly what is to be done and creating a drone army (which could be replaced by outsourcing or automation), try setting clear expectations of the outcomes you want achieved and let the team go for it. This will give the team a sense of worth as you are touching on Maslow's Hierarchy of Needs by giving them a channel for creativity and problem solving, but probably more importantly a sense of respect and belonging. Obviously it doesn't have to be a free-for-all, which would no doubt result in total failure, but simple things like burn down charts, daily stand-up's can give 'management' visibility of progress.

So what can you do if you are not management?

I believe that people are attracted to like people. If you value and exercise principles like mastery, purpose, independence, interdependence then I think you will attract to companies where that is the culture. Easiest thing to do is trying for mastery. In the information age it is simple to learn new skills. Simple is not the same as Easy however. Finding the information on a subject is not the same as mastering it, so you have to jump in feet first and make mistakes. Ideally this is not on mission critical projects. I do a lot of mini projects at home to learn a new skill. Some companies like Google are now offering time at work to do this stuff. If your company does not offer this time, then you may want to consider your priorities and see if you can make time for self development. 30 minutes on the way to work, you can read 10 pages of a book. That should allow you around 1 book per month. If you drive/cycle to work, then maybe audio books or podcasts are for you. This is only one part of the path to mastery. You then need to distil the information you are collecting. At the end of a chapter, jot down some notes in your own words of what you just learnt. The act of distilling the information will help you understand it better. To further improve your understanding, actually do something that requires your new skill. Lastly to really see if you understand the skill, try teaching someone else the skill. This really shows you where you are weak still.

Next, start automating the prescriptive parts of your role. For C# guys this mainly done for us with tools like automated build servers and Resharper. If have not worn out our tab key then as a C# code you probably don't use snippets/live templates enough. If you don't have a little keyboard short cut for getting latest, building, running tests and checking in then automate it! If part of your role can't be automated, try outsourcing it. It is amazing what people will do if you ask them nicely - "Excuse me Tony you don't mind taking this mail to the mail room for me?"

Once you are on the track for mastery by creating some sort of self development routine*, you have automated or outsourced the prescriptive parts to your role, you can now concentrate on delivering value from your role. This is where purpose comes in. Ideally as you develop more skills, reduce drone work and start delivering more value from your role management may feel that you can have a level of autonomy. From their perspective giving you autonomy now is really just reducing their work load. If you constantly deliver value, management will naturally move away from compliance management to engagement.

S.M.A.R.T. goals

Projects – Measuring success and providing boundaries

*I try to read a minimum one book per month. I also try to keep up with my blog subscriptions but constantly have hundreds of unread post. As a guide apparently Bill Gates reads 3 books a weekend!

How not to write an Interface in C# 3.0

2009-08-14T23:05:00.001+01:00

While working with the IUnityContainer interface from Microsoft's Patterns and Practices team, I decided that it was well worth a post on how not to design interfaces. Recent discussion amongst Rhys and Colin (here as well) have been interesting but I imagine most would agree that both arguments are probably fighting over which polish to use on their code. If these are the big battles you have to face at work then I am jealous.

Introducing IUnityContainer ....
42 members are defined on this interface. Fail.
With much restraint I wont go on about it and flame the guys who write it. Most people know of my disdain for the P&P team at Microsoft. So what tips do I give to make this a usable again? Lets break down the solution into a few tips:

Extension methods
Cohesion
Intention revealing interfaces

How do each of these help us. Let look at some code I was writing pre-extension methods. It was a validation interface that had 2 methods, Validate and Demand

public interface IValidator<T>
{
  IEnumerable<ValidationErrors> Validate(T item);

  ///<exception cref="ValidationException">Thrown when the item is not valid</exception>
  void Demand(T item);
}

The problem with this interface is that all implementations of the interface would/could implement Demand the same way; Make a call to Validate(T item) and if any ValidationErrors came back throw an exception with the validation failures. Time ticks by and I realise the now that I have extension methods in C# 3.0 (.NET 3.5). I only need to have the Validate method on my interface now and provide the Demand implementation as an extension method. The code now becomes:

public interface IValidation<T>
{
  /// <summary>
  /// Validates the specified item, returning a collection of validation errors.
  /// </summary>
  /// <param name="item">The item to validate.</param>
  /// <returns>Returns a <see cref="T:ArtemisWest.ValidationErrorCollection"/> that is empty for a valid item, 
  /// or contains the validation errors if the item is not in a valid state.</returns>
  /// <remarks>
  /// Implementations of <see cref="T:ArtemisWest.IValidation`1"/> should never return null from the validate method.
  /// If no validation errors occured then return an empty collection.
  /// </remarks>
  IEnumerable<ValidationErrors> Validate(T item);
}
public static class ValidatorExtensions
{
  /// <summary>
  /// Raises a <see cref="T:ArtemisWest.ValidationException"/> if the <paramref name="item"/> is not valid.
  /// </summary>
  /// <typeparam name="T">The type that the instance of <see cref="T:IValidation`1"/> targets.</typeparam>
  /// <param name="validator">The validator.</param>
  /// <param name="item">The item to be validated.</param>
  /// <exception cref="T:ArtemisWest.ValidationException">
  /// Thrown when validating the <paramref name="item"/> returns any errors. Only the first 
  /// validation error is raised with the exception. Any validation errors that are marked as
  /// warnings are ignored.
  /// </exception>
  public static void Demand<T>(this IValidation<T> validator, T item)
  {
    foreach (var error in validator.Validate(item))
    {
      if (!error.IsWarning)
      {
        throw new ValidationException(error);
      }
    }
  }
}

Then end result is that the API feels the same as I have access to both methods, but the cost of implementing my interface is reduced to just its core concern.
So, extension methods is one trick we have in the bag. Next cohesion.

The recent discussion between Rhys and Colin is "how many members belong on an interface?" I think both will agree that answer is not 42. Juval Lowy made a great presentation at TechEd2008 on Interfaces and that we should be aiming for 3-7 members per interface. This provides us with a level of cohesion and a low level of coupling. Lets look at some of the members on the IUnityContainer Interface.

8 overloads of RegisterInstance
16 overloads of RegisterType
Add/Remove "Extensions" methods
4 overloads of BuildUp
2 overloads of ConfigureContainer
CreateChildContainer
4 overloads of Resolve
2 overloads of ResolveAll
A TearDown method
and a Parent property

Whew! Well how can we tame this beast? When I look at this interface I see certain groups that look like they are related in usage. They would be the

Register and Resolve functionality
And and Remove extensions functionality
Build up and teardown functionality
Container hierarchy functionality (Parent and CreateChildContainer)
Container configuration

Interestingly on our current project we only use the Register/Resolve functionality.
These five groups to me have some level of cohesion. That to me then makes them a candidate for there own interfaces. The big giveaway being that I use unity quite successfully but never use 4/5 of the functionality defined on this interface. So our 2nd tip would be to split out these groups on functionality into there own interfaces.
But what do we name the new interfaces? This is our 3rd tip:

Intention revealing interfaces.

Looking at my list I would imagine some useful names could be:

IContainer
IExtensionContainer
ILifecycleContainer
INestedContainer
IConfigurableContainer

To be honest, I have put little thought into these names. Normally I would put a LOT of effort into getting the naming right, but I don't work on the P&P team so these changes will never be done so why waste my time? Edit: My laziness here really does take the wind our of the sails of this argument. Sorry.
Ok so how can we bring this all together? My proposal would be to have 6 interfaces

IContainer
IExtensionContainer
ILifecycleContainer
INestedContainer
IConfigurableContainer
IUnityContainer : IContainer, IExtensionContainer, ILifecycleContainer, INestedContainer, IConfigurableContainer

Next I would create some extension methods to deal with the silly amount of duplication the multiple overloads incur. Looking at the implementation in UnityContainerBase I would think that all of these methods could be candidates for extension methods

public abstract class UnityContainerBase : IUnityContainer, IDisposable
{
  //prior code removed for brevity
  public IUnityContainer RegisterInstance<TInterface>(TInterface instance)
  {
    return this.RegisterInstance(typeof(TInterface), null, instance, new ContainerControlledLifetimeManager());
  }
  public IUnityContainer RegisterInstance<TInterface>(string name, TInterface instance)
  {
    return this.RegisterInstance(typeof(TInterface), name, instance, new ContainerControlledLifetimeManager());
  }
  public IUnityContainer RegisterInstance<TInterface>(TInterface instance, LifetimeManager lifetimeManager)
  {
    return this.RegisterInstance(typeof(TInterface), null, instance, lifetimeManager);
  }
  public IUnityContainer RegisterInstance(Type t, object instance)
  {
    return this.RegisterInstance(t, null, instance, new ContainerControlledLifetimeManager());
  }
  public IUnityContainer RegisterInstance<TInterface>(string name, TInterface instance, LifetimeManager lifetimeManager)
  {
    return this.RegisterInstance(typeof(TInterface), name, instance, lifetimeManager);
  }
  public IUnityContainer RegisterInstance(Type t, object instance, LifetimeManager lifetimeManager)
  {
    return this.RegisterInstance(t, null, instance, lifetimeManager);
  }
  public IUnityContainer RegisterInstance(Type t, string name, object instance)
  {
    return this.RegisterInstance(t, name, instance, new ContainerControlledLifetimeManager());
  }
  //Remaining code removed for brevity
}

all of these methods just delegate to the one method overload left as abstract

public abstract IUnityContainer RegisterInstance(
    Type t, 
    string name, 
    object instance, 
    LifetimeManager lifetime);

The obvious thing to do here is to just make all of these extension methods in the same namespace as the interfaces

public static class IContainerExtensions
{
  public IContainer RegisterInstance<TInterface>(this IContainer container, TInterface instance)
  {
    return container.RegisterInstance(typeof(TInterface), null, instance, new ContainerControlledLifetimeManager());
  }
  public IContainer RegisterInstance<TInterface>(this IContainer container, string name, TInterface instance)
  {
    return container.RegisterInstance(typeof(TInterface), name, instance, new ContainerControlledLifetimeManager());
  }
  public IContainer RegisterInstance<TInterface>(this IContainer container, TInterface instance, LifetimeManager lifetimeManager)
  {
    return container.RegisterInstance(typeof(TInterface), null, instance, lifetimeManager);
  }
  public IContainer RegisterInstance(this IContainer container, Type t, object instance)
  {
    return container.RegisterInstance(t, null, instance, new ContainerControlledLifetimeManager());
  }
  public IContainer RegisterInstance<TInterface>(this IContainer container, string name, TInterface instance, LifetimeManager lifetimeManager)
  {
    return container.RegisterInstance(typeof(TInterface), name, instance, lifetimeManager);
  }
  public IContainer RegisterInstance(this IContainer container, Type t, object instance, LifetimeManager lifetimeManager)
  {
    return container.RegisterInstance(t, null, instance, lifetimeManager);
  }
  public IContainer RegisterInstance(this IContainer container, Type t, string name, object instance)
  {
    return container.RegisterInstance(t, name, instance, new ContainerControlledLifetimeManager());
  }
}

This now reduces our IContainer Interface to just the one method

public interface IContainer
{
  IContainer RegisterInstance(Type t, string name, object instance, LifetimeManager lifetime);
}

One thing to note here is that we have broken the contract because we now return IContainer not IUnityContainer. We will come back to this problem later.
If we then follow suit on the other interfaces we have created, we will have 6 interfaces that look like this:

public interface IContainer
{
  IContainer RegisterType(Type from, Type to, string name, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers);
  IContainer RegisterInstance(Type t, string name, object instance, LifetimeManager lifetime);
  object Resolve(Type t, string name);
  IEnumerable<object> ResolveAll(Type t);
}
public interface IExtensionContainer
{
  IExtensionContainer AddExtension(UnityContainerExtension extension);
  IExtensionContainer RemoveAllExtensions();
}
public interface ILifecycleContainer
{
  object BuildUp(Type t, object existing, string name);
  void Teardown(object o);
}
public interface INestedContainer
{
  INestedContainer CreateChildContainer();
  INestedContainer Parent { get; }
}
public interface IConfigurableContainer
{
  object Configure(Type configurationInterface);
}
public interface IUnityContainer : IDisposable, IContainer, IExtensionContainer, ILifecycleContainer, INestedContainer, IConfigurableContainer
{}

So now we have some much more managable interfaces to work with. However, we have broken the feature that it had before of returning IUnityContainer from each method. You may ask why would you return the instance when clearly you already have the instance? By doing so you can create a fluent interface. See my other post on Fluent interfaces and DSLs for more information.
Now that we have removed all the noise from the interfaces we actually have a reasonable number of members to work with. This makes me think that maybe we can refactor back to a single interface? Lets have a look:

public interface IUnityContainer : IDisposable
{
  IUnityContainer RegisterType(Type from, Type to, string name, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers);
  IUnityContainer RegisterInstance(Type t, string name, object instance, LifetimeManager lifetime);
  object Resolve(Type t, string name);
  IEnumerable<object> ResolveAll(Type t);
  IUnityContainer AddExtension(UnityContainerExtension extension);
  IUnityContainer RemoveAllExtensions();
  object BuildUp(Type t, object existing, string name);
  void Teardown(object o);
  IUnityContainer Parent { get; }
  IUnityContainer CreateChildContainer();
  object Configure(Type configurationInterface);
}

Well that is 13 members, which is above my happy limit of 10 and nearly double my ideal limit of 7, however...I think this would be a vast improvement on the silly interface that currently exists with its 42 members.
Just for fun here are the Extension methods that would complete the interface to bring it back to feature complete.

public static class IUnityContainerExtentions
{
  public static IUnityContainer AddNewExtension<TExtension>(this IUnityContainer container) where TExtension : UnityContainerExtension, new()
  {
    return container.AddExtension(Activator.CreateInstance<TExtension>());
  }
  public static T BuildUp<T>(this IUnityContainer container, T existing)
  {
    return (T)container.BuildUp(typeof(T), existing, null);
  }
  public static object BuildUp(this IUnityContainer container, Type t, object existing)
  {
    return container.BuildUp(t, existing, null);
  }
  public static T BuildUp<T>(this IUnityContainer container, T existing, string name)
  {
    return (T)container.BuildUp(typeof(T), existing, name);
  }
  public static TConfigurator Configure<TConfigurator>(this IUnityContainer container) where TConfigurator : IUnityContainerExtensionConfigurator
  {
    return (TConfigurator)container.Configure(typeof(TConfigurator));
  }
  public static IUnityContainer RegisterInstance<TInterface>(this IUnityContainer container, TInterface instance)
  {
    return container.RegisterInstance(typeof(TInterface), null, instance, new ContainerControlledLifetimeManager());
  }
  public static IUnityContainer RegisterInstance<TInterface>(this IUnityContainer container, string name, TInterface instance)
  {
    return container.RegisterInstance(typeof(TInterface), name, instance, new ContainerControlledLifetimeManager());
  }
  public static IUnityContainer RegisterInstance<TInterface>(this IUnityContainer container, TInterface instance, LifetimeManager lifetimeManager)
  {
    return container.RegisterInstance(typeof(TInterface), null, instance, lifetimeManager);
  }
  public static IUnityContainer RegisterInstance(this IUnityContainer container, Type t, object instance)
  {
    return container.RegisterInstance(t, null, instance, new ContainerControlledLifetimeManager());
  }
  public static IUnityContainer RegisterInstance<TInterface>(this IUnityContainer container, string name, TInterface instance, LifetimeManager lifetimeManager)
  {
    return container.RegisterInstance(typeof(TInterface), name, instance, lifetimeManager);
  }
  public static IUnityContainer RegisterInstance(this IUnityContainer container, Type t, object instance, LifetimeManager lifetimeManager)
  {
    return container.RegisterInstance(t, null, instance, lifetimeManager);
  }
  public static IUnityContainer RegisterInstance(this IUnityContainer container, Type t, string name, object instance)
  {
    return container.RegisterInstance(t, name, instance, new ContainerControlledLifetimeManager());
  }
  public static IUnityContainer RegisterType<T>(this IUnityContainer container, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(typeof(T), null, null, null, injectionMembers);
  }
  public static IUnityContainer RegisterType<TFrom, TTo>(this IUnityContainer container, params InjectionMember[] injectionMembers) where TTo : TFrom
  {
    return container.RegisterType(typeof(TFrom), typeof(TTo), null, null, injectionMembers);
  }
  public static IUnityContainer RegisterType<T>(this IUnityContainer container, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(typeof(T), null, null, lifetimeManager, injectionMembers);
  }
  public static IUnityContainer RegisterType<TFrom, TTo>(this IUnityContainer container, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers) where TTo : TFrom
  {
    return container.RegisterType(typeof(TFrom), typeof(TTo), null, lifetimeManager, injectionMembers);
  }
  public static IUnityContainer RegisterType<T>(this IUnityContainer container, string name, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(typeof(T), null, name, null, injectionMembers);
  }
  public static IUnityContainer RegisterType<TFrom, TTo>(this IUnityContainer container, string name, params InjectionMember[] injectionMembers) where TTo : TFrom
  {
    return container.RegisterType(typeof(TFrom), typeof(TTo), name, null, injectionMembers);
  }
  public static IUnityContainer RegisterType(this IUnityContainer container, Type t, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(t, null, null, null, injectionMembers);
  }
  public static IUnityContainer RegisterType<T>(this IUnityContainer container, string name, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(typeof(T), null, name, lifetimeManager, injectionMembers);
  }
  public static IUnityContainer RegisterType<TFrom, TTo>(this IUnityContainer container, string name, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers) where TTo : TFrom
  {
    return container.RegisterType(typeof(TFrom), typeof(TTo), name, lifetimeManager, injectionMembers);
  }
  public static IUnityContainer RegisterType(this IUnityContainer container, Type t, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(t, null, null, lifetimeManager, injectionMembers);
  }
  public static IUnityContainer RegisterType(this IUnityContainer container, Type t, string name, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(t, null, name, null, injectionMembers);
  }
  public static IUnityContainer RegisterType(this IUnityContainer container, Type from, Type to, params InjectionMember[] injectionMembers)
  {
    container.RegisterType(from, to, null, null, injectionMembers);
    return container;
  }
  public static IUnityContainer RegisterType(this IUnityContainer container, Type t, string name, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(t, null, name, lifetimeManager, injectionMembers);
  }
  public static IUnityContainer RegisterType(this IUnityContainer container, Type from, Type to, LifetimeManager lifetimeManager, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(from, to, null, lifetimeManager, injectionMembers);
  }
  public static IUnityContainer RegisterType(this IUnityContainer container, Type from, Type to, string name, params InjectionMember[] injectionMembers)
  {
    return container.RegisterType(from, to, name, null, injectionMembers);
  }
  public static T Resolve<T>(this IUnityContainer container)
  {
    return (T)container.Resolve(typeof(T));
  }
  public static T Resolve<T>(this IUnityContainer container, string name)
  {
    return (T)container.Resolve(typeof(T), name);
  }
  public static object Resolve(this IUnityContainer container, Type t)
  {
    return container.Resolve(t, null);
  }
  public static IEnumerable<T> ResolveAll<T>(this IUnityContainer container)
  {
    //return new <ResolveAll>d__0<T>(-2) { <>4__this = this };
    //This implementation requires more effort than I am willing to give (6hours till my holiday!)
  }
}

Bloody hell. Imagine trying to implement that on every class that implemented the interface!

While I know this entire post is academic as we cant change Unity, but I hope that sews some seeds for other developers so that when they design their next interface it wont have silly overloads that just make the consuming developer's job harder than it ought to be.

Code inspections

2009-07-26T02:00:00.001+01:00

Code Inspections (or Code Reviews or Peer Reviews) is a technique a team can implement to improve the quality of the software they deliver. It is initially a bit of a cultural shift for many teams but can produce benefits not just in software but if done well also in team morale.

Goals

The goals for performing a Code Inspection are the following:

To cross pollinate Domain knowledge
To cross pollinate technical knowledge
To maintain a consistent team process
To identify any obvious errors

A code inspection should be an enjoyable experience where two developers are able to bond and cooperatively improve the quality of software delivered. The exercise should not feel like a witch-hunt or a boasting exercise for one developer to flex their ability over another. Initially the exercise may feel somewhat uncomfortable for the team, this is usually due to many different styles and tolerances the each member of the team brings to the collective skill set. In these scenarios it is better to agree amongst the team the level of detail that is in the remit of the inspections.

Introducing a team to Code Inspections

It is probably best if a team ease into code inspections, for it is better to be doing some code inspections and forgiving style differences than it is to be too aggressive in initial code inspections and cause tension in the team which sees the exercise become avoided. While this may be difficult to accept initially, one must remember that increasing the quality of today's code is of less importance than spreading the knowledge through out the team and maintain a quality team spirit. To know what to target in initial inspections, the team should conduct a brief team meeting and pick a handful of areas that the team agree needs work. This way we can get the biggest gain without creating to much friction.

Tools

The "what" to look for is of less importance than the process itself but the process can be aided quite nicely by the use of automated tools. In the .NET community the tools I would recommend are FxCop (or the Visual Studio Code Analysis), Simian (identifies duplicate code) and NCover (for test coverage).
To ease into code inspections I think it is best to set the bar quite low. For FxCop just enable the following Rules

Design
Naming
Performance
Security
Usage

While some teams may have their own style that might conflict with some of the styles in FxCop, ultimately this is a community standard now. As developers move between jobs it is great to eliminate this kind of induction friction when all .NET developers can be "singing from the same hymn sheet".
Simian provides the ability to set a tolerance of duplication to 6 lines of duplicate code. This should be fine.
Assuming the team is TDD or at least writing tests the the bar for test coverage could start at something like 80%. While I understand that chasing 100% test coverage can be frowned upon, it is nice to know that code was writing for a reason and does what it was intended on doing.

Setting expectations

Once the initial team meeting has been completed a set of expectations should be drawn up. This will be the bench mark for code inspections. The benchmark should be set at a level that everyone in the team agrees that they are happy for their code to be "rejected" if it does not meet the expectations. As you can imagine this in itself will lower the bar quite considerably as developers know that there are circumstances beyond their control which may prohibit them meeting the expectations that they would like to meet. Project deadlines, new technologies and processes can affect a developers ability to meet these goals. Remember the goals for code inspection is not perfect code, it is to improve the ability of the team to create better code, and continually get better at creating better code.
An example for an initial code inspection check list may look like this for a C# 3.0 project:

Exceptions that are explicitly thrown from a public or protected members must be documented with XML comments.
FxCop Naming and Performance rules must be met.
Classes should be testable.

While this is a very small list of things to consider, just setting these expectations combined with knowledge sharing that the team would perform would see a dramatic improvement on any team not already doing code inspections.
Once the team is comfortable with the process and culture of code inspections, have another meeting to raise the bar a little bit. Again, refrain from being too eager. All team member should feel comfortable with the expectations being set.

Performing Code Inspections

Be fore we cover the actual actions of what to do when doing a code inspection just quickly remeber the goals of a code inspection

To cross pollinate Domain knowledge
To cross pollinate technical knowledge
To maintain a consistent team process
To identify any obvious errors

The inspection is not to create perfect code.
Steps to perform a code inspection (the first few step are as a courtesy to the reviewer):

First perform a quick review yourself.
Indicate to the person you would like to review your code that you will want some of their time in a few moments. It is important that this person not be the same few people. Have a method where you cycle through the team (clock wise round the office, youngest to oldest etc). Generally you will want to do most of your reviews with your immediate team, but if you have a larger development team/department that can be involved in the inspection be sure to include them too. *
Perform any automated analysis (FxCop, Simian, Unit tests etc...)
Open all the necessary documents ready to go, and consider the logical path to show the reviewer through your code.
Some prefer to review code on paper. This is very good for removing the desire to solve any problems with the code, it also allows you to easily make notes against the code and possibly perform the review in a quiet place. However, I also think that if done properly this creates a mass of wasted paper and ink.
When both parties are ready, have the author describe the intent of the code to the reviewer. This should start with a 10 second overview of what set of code does.
Show the reviewer the output of any automated analysis (FxCop, Tests, Coverage etc.). Obviously the results of which should met the set expectations.
If at the computer give the reviewer the mouse and keyboard. If using a print out have the print out in front of the reviewer.
Describe what the code does, ideally from entry point to each of the leaves. Very good code will basically read like prose anyway so this would be very easy.
The reviewer sets the pace of the review by scrolling or navigating to classes/methods as they understand the code.
The reviewer prompts for "what-if" scenarios and bookmarks any concerns (Ctrl+k+k in Visual Studio). No attempt is made to "fix" any code by either party.
As the reviewer bookmarks any code they should also take a note of why the bookmark was made (Unclear code, not meeting agreed expectations, possible error etc). This can be done on paper or if available a Code Inspection prompt sheet.
The author generally will identify most of their mistakes as they describe the code to the reviewer.
If the code has been flagged in any way that would fail to meet the set expectations -or- would result in probably faulty software; the Author requests the the Reviewer come back to verify their changes and approve the code.**
Once the review(s) have come to a point where no probable faults exists in the code and the team expectations are met the code review is complete.

*In my experience my project teams have been generally in sizes of 3-8. In these scenarios I would try to have each member of the team review my code over the course of a week. Once I had engaged each team member I would next engage someone from another team. This allows for a knowledge transfers across teams too (like finding out each team is using different logging tools!).

**By having the Author suggest that they fix their code, this creates far less stress for the reviewer to feel as if they are judging the authors code. Essentially the reviewer is a soundboard and prompt for the author.

Hints and Tips

Ensure that the team is clear that both the Reviewer and the Author are there to find any defects. It is a cooperative approach.
Focus on finding defects and failed expectations, not fixing them.
The biggest benefit for performing Code Inspections is that it creates a culture where the team all get better technically and also domain knowledge is not isolated to pockets of people.
Inspect all new and modified code
For best results review often and review early. Not many developers write so much code that 4 reviews in a day would be necessary, but not may developers would write so little code that only 1 or no code reviews would be done in a day. An exmple of a set of code to review could be when an MVP screen is complete; review the Model, View, Presenter and tests. I imagine few developer produce more than 4 tested screens per day.
As you get better at code reviews they should become less and less formal and quicker to execute. Extreme Programming takes peer review to the next level by have 2 developer work together constantly in a fashion called Pair Programming.
Some teams keep track of their code reviews an can create statistics to analyse where current weaknesses are. If you see value in this than give it a go, but make sure it doesn't just create "busy work". I believe that if the inspections are done correctly the team will know that there is an area of need and should raise it appropriately.
Code reviews should be done at the time the code was completed. New code should not be started until the current code has been inspected and approved. Keeping thing fresh in your head is key to a good review. It also means that "unfinished"/un-inspected code doesn't just slip out into production.
Some source/version control systems allow you to attribute check ins. If this is the case then code that has passed a check in can be attributed ideally with the reviewer's details. This should then allow the team to analyse all code that has not been reviewed (ie. last check-in for a file was not attributed).

.NET Parallel Framework - Videos

2009-07-09T23:12:00.001+01:00

Way back in June 2008 Joe Albahari gave a presentation on the upcoming parallel programming features in .NET 4.0. I have finally published the recording of his session here. Just for reference the Parallel Framework encompasses PLINQ and is sometimes referred to as the TPL (Task Parallel Library) or System.Threading.dll

Sorry about the delay in the post.

Joe Albarhari Parallel FX and PLINQ videos : 6 videos @ ~10min and 40-50MB each.

Responsive WPF User Interfaces Part 7

2009-06-06T00:26:00.001+01:00

Creating a responsive Image control

In part 6 we discovered problems even when we follow the guidelines from this series. In this case we find that sometimes it can be the controls themselves that are the cause of the lagging interface. In part 6 it was the Image control that was to blame. I make the assumption that it was trying to perform it's bitmap decoding on my precious dispatcher thread. However not only Image controls can get ugly, Charles Petzold shows some problems you may find when using chart controls (or any Items Control) in his Writing More Efficient Items Controls article.

The first thing I wanted to do was to just sub class Image however, I personally was stumped as to how I would then create my control template for it. Next I actually only wanted to accept URIs as my source so that I can do the decoding from file to an ImageSource myself explicitly and not on the UI thread via a TypeConverter. So I want a sub class of Control with a dependency property of UriSource and then 3 read-only properties ImageSource, IsLoading and HasLoadFailed. UriSource will provide the hook to bind your file name to. ImageSource will then provide the decoded ImageSource to be displayed. IsLoading and HasLoadFailed will be there so that you can update the UI appropriate to the lifecycle of the image.

The Style and control template I came up for this design was this:

<Style TargetType="{x:Type controls:Thumbnail}">
  <Setter Property="Template">
    <Setter.Value>
      <ControlTemplate TargetType="{x:Type controls:Thumbnail}">
        <Image x:Name="ImageThumbnail"
               Source="{TemplateBinding Image}"
               HorizontalAlignment="{TemplateBinding HorizontalAlignment}"
               VerticalAlignment="{TemplateBinding VerticalAlignment}"
               MaxHeight="{TemplateBinding MaxHeight}"
               MaxWidth="{TemplateBinding MaxWidth}"
               Stretch="Uniform"
               StretchDirection="DownOnly" />
      </ControlTemplate>
    </Setter.Value>
  </Setter>
</Style>

Where I want to use the Thumbnail control I can have some XAML that hides and shows a loading indicator. This example here just shows the text "Loading Image..." but would probably have a nice animation or an indeterminate progress indicator.

<Border CornerRadius="4" BorderBrush="Silver" BorderThickness="1"
        Width="150"
        Height="150"
        Margin="10">
  <Grid>
    <local:Thumbnail x:Name="Thumbnail"
                     MaxWidth="148"
                     MaxHeight="148"
                     UriSource="{Binding}" 
                     HorizontalAlignment="Center" VerticalAlignment="Center"
                     ToolTip="{Binding}">
      <local:Thumbnail.Style>
        <Style TargetType="{x:Type local:Thumbnail}">
          <Setter Property="Visibility" Value="Visible"/>
          <Style.Triggers>
            <DataTrigger Binding="{Binding RelativeSource={RelativeSource Self}, Path=IsLoading}" Value="True">
              <Setter Property="Visibility" Value="Collapsed"/>
            </DataTrigger>
          </Style.Triggers>
        </Style>
      </local:Thumbnail.Style>
    </local:Thumbnail>

    <TextBlock Text="Loading image..."
               Foreground="Gray"
               HorizontalAlignment="Center"
               VerticalAlignment="Center">
      <TextBlock.Style>
          <Style TargetType="TextBlock">
              <Setter Property="Visibility" Value="Visible"/>
              <Style.Triggers>
                  <DataTrigger Binding="{Binding ElementName=Thumbnail, Path=IsLoading}" Value="False">
                      <Setter Property="Visibility" Value="Collapsed"/>
                  </DataTrigger>
              </Style.Triggers>    
          </Style>
      </TextBlock.Style>
    </TextBlock>
  </Grid>
</Border>

Well that is the easy bit, the public API. Trust me it gets more interesting.

My gut feeling was basically take the code from reflector and use that to decode Uris to ImageSource objects, but perform the action on a background thread. Two problems here:

Most of the fun code that happens when the URI is being parsed and then decoded, is internal. :-(
You cant pass most sub-classes of ImageSource across threads. If you create them on one thread they cant be accessed on another thread. So this stops us blindly copying code from reflector as all that code only runs on the dispatcher thread so wont face this problem.

We do have something to work with however: WriteableBitmap.

In some vain attempt at brevity (he says in his 7th post in the series!), I will try to skip over all of the brick walls I faced and jump straight to the solution I came up with. This was a great learning experience for me, and in the spirit of a learning experience the code is fairly rough (with TODO comments still intact). I will try to build this control into a stable control but in its current state is very much demo-ware.

First thing I want to achieve was to have the decoding and if possible the reading from disk happen off the UI thread. Lets start with the easy bit; reading the file into memory. I'm going to keep that simple and just go with grabbing the file as a byte array like this

byte[] buffer = File.ReadAllBytes(UriSource);

Next I want to decode the byte array into some form of ImageSource. To do this I have used a combination of the BitmapDecoder and the WriteableBitmap. First I take the byte array and load it into a MemoryStream, and then pass the stream to the BitmapDecoder factory method Create. This will return me an instance of one of its implementations (BitmapDecoder is abstract/MustInherit). From here I make a bold assumption that we only care about the first "Frame" of the image. I believe that most formats don't support multiple frames but formats like GIF do which allows them to have animation features. Anyway, in my demo-ware code I just take the first frame.

using (Stream mem = new MemoryStream(buffer))
{
  BitmapDecoder imgDecoder = null;
  imgDecoder = BitmapDecoder.Create(mem, BitmapCreateOptions.None, BitmapCacheOption.None);

  BitmapFrame frame = imgDecoder.Frames[0];
  double scale = GetTransormScale(maxWidth, maxHeight, frame.PixelWidth, frame.PixelHeight);

  BitmapSource thumbnail = ScaleBitmap(frame, scale);

  // this will disconnect the stream from the image completely ...
  var writable = new WriteableBitmap(thumbnail);
  writable.Freeze();
  return writable;
}

From here I figure out the ratio that I want to scale it to. There doesn't seem much point in returning an 8MB image if we only want to see it as 300x300 does it? Next we request the image as what is almost our final product. We have a helper method scale the frame and return it as a BitmapSource. We can't assign this BitmapSource back to our ImageSource dependency property as they don't play nice over thread boundaries. So, the last thing we need to do is take the BitmapSource we just generated and create a WriteableBitmap from it and then call its Freeze method. This puts the WriteableBitmap in an immutable state which then makes it thread safe. Whew!

Just for reference here is the GetTransformScale method

private static double GetTransormScale(double maxWidth, double maxHeight, double currentWidth, double currentHeight)
{
  double xRatio = maxWidth / currentWidth;
  double yRatio = maxHeight / currentHeight;
  double resizeRatio = (xRatio > yRatio) ? xRatio : yRatio;
  if (resizeRatio > 1)
    resizeRatio = 1;
  return resizeRatio;
}

and the ScaleBitmap method

private static BitmapSource ScaleBitmap(BitmapSource source, double scale)
{
  if (scale > 0.9999 && scale < 1.0001)
  {
    return source;
  }
  var thumbnail = new TransformedBitmap();
  thumbnail.BeginInit();
  thumbnail.Source = source;
  var transformGroup = new TransformGroup();
  transformGroup.Children.Add(new ScaleTransform(scale, scale));
  thumbnail.Transform = transformGroup;
  thumbnail.EndInit();
  return thumbnail;
}

So all things considered, once you get over the decoding stuff and then wrestle with the various subclasses of ImageSource that play their part, its not too bad. But like I said earlier; it gets more interesting. So far we have only looked at how to decode the image, we have yet to consider how to make the call to perform the decoding. My first instinct was to make the call on any change to the UriSource property. However I may not have the MaxHeight and MaxWidth information at that point in time. This caused me much stress over when should I call this decode functionality. If no value is ever going to be set for MaxHeight or MaxWidth then I should just process the image, but if first the UriSource is set then the MaxHeight, then the MaxWidth I would end up creating 3 asynchronous calls to render 3 different sized images. This would be a disaster as it would surely end up with race conditions and most likely the wrong sized image being displayed. The other obvious problem with that is that we would be performing 3 times the work. Hmm.

My solution (and mileage may vary as I am green to concurrent programming models) was to create a stack of render requests for each instance of a Thumbnail. As a property was set then a request would be added to the stack and a request to start processing the stack would occur. Periodically while processing the image I would check to see if the current work was invalidated by any new requests to the stack. If the current request was invalid it would terminate its work. The request to start processing the stack would simply try to pop the last request from the stack and clear out all other requests (effectively ignoring stale requests). On returning from the render request it would loop back and try to pop anything new from the stack. This last part while important is very much implementation details and could vary dramatically from anything you may implement. I have just read over the code myself and I could do with some work. It is amazing what just 3months (and reading Joe Duffy's Concurrent Programming for Windows) does for your appreciation of your code.

As always the code is available for you to have a play. Open it up, pick it to pieces, use what you like. Obviously I take no responsibility for the code if you do choose to use it as this is intended on being a learning exercise. Having said that I do try to produce good code for my demos so if you do see something that is not good enough let me know.

This has been a fun series and I hope you liked it and learnt as much as I did from it.

Back to series Table Of Contents

Working version of the code can be found here

Responsive WPF User Interfaces Part 6

2009-05-26T23:43:00.001+01:00

Unresponsive Controls

Sorry about the unreasonable delay between this post and the last post in this series. Interestingly readers that have been following the series have already jumped ahead to see that not all is fine even if you follow the guidance provided. 'Arne' makes a comment on the last post that he is having some problems even when he follows the guidance especially when loading a lot of images. Some of the team I work with are currently working with graphing/charting products that are great with tens and hundreds of rows of data but create an unresponsive UI when pushed to use thousands of rows of data (eg +5 years of daily stock info).

So if we are following the rules what can we do? Well let us look at a concrete problem first and for consistency let's stay with the PhotoAlbum project. For a quick refresher, we have a "ViewModel" with a property Images that is of type ObservableCollection<Uri>. The XAML snippet that displays the Images looks like this:

<ListBox ItemsSource="{Binding Images}"
         ScrollViewer.HorizontalScrollBarVisibility="Disabled">
  <ListBox.ItemTemplate>
    <DataTemplate>
      <Border Background="AliceBlue"
              CornerRadius="2"
              Width="150"
              Height="150">
        <Image Source="{Binding}"
               MaxHeight="148"
               MaxWidth="148"
               Stretch="Uniform"
               StretchDirection="DownOnly"
               HorizontalAlignment="Center"
               VerticalAlignment="Center"
               ToolTip="{Binding}" />
      </Border>
    </DataTemplate>
  </ListBox.ItemTemplate>
  <ListBox.ItemsPanel>
    <ItemsPanelTemplate>
      <WrapPanel IsItemsHost="True" />
    </ItemsPanelTemplate>
  </ListBox.ItemsPanel>
</ListBox>

Now in theory there is nothing wrong with this XAML. From a declarative programming point of view, I think it is well written (Ed: Cause I wrote it!) as it describes what we want (Items in the ListBox to be wrapped horizontally, displayed as Images within a border.

The problem with this declarative code is what most people fear with declarative code: what is actually happening under the hood? Before we trek too far, if I run the code (see example 6_UnresponsiveImages in the code example) I can select folders with small amount of small images just fine. The application runs ok. The current code becomes a problem when I point it to folders with large images. This is compounded when there are lots of those large images or when I am running on a slower/older computer.

We will quickly go off track to consider the likelihood of the large image slow computer problem. Digital cameras currently are getting cheaper and the quality is going up. Mid range cameras range between 8-12 megapixel and can have in excess of 8GB of memory. Phones now are coming out with 5-8 megapixel cameras. If you come back and read this 12months from now, I guess you will laugh at the small sizes I refer to. Next consider the average user's computer. Rightly or wrongly, the release of vista saw very few new features that users felt they needed. Many people still run Win XP and do so on old hardware. Pentium 4 chips are still pervasive in many homes and offices. Loading a gig of 10 megapixel images in our current app does prove to be a performance problem.

We actually have 2 problems here:

loading of the image in it's full resolution
loading the image on the UI thread

We will only discuss issue 2 because technically this series is about responsive UIs, not performance (similar but not the same problem). So to get to the bottom of why our UI is laggy even though we have

loaded our Images on a different thread,
updated the UI using the dispatcher
and written nice XAML

we should understand a little bit about the Image control. The Image control has a Source property. Most often this source property is given a string or Uri, however some may have noticed that it actually is of type ImageSource. Through the magic of TypeConverters our strings and URI are converted to an appropriate sub class of ImageSource, most likely BitmapFrame. From having a quick browse over the code implemented around the ImageSource, BitmapSource, BitmapFrame & BitmapDecoder, I can only deduct that the only code executed off the UI thread is the downloading of the image where the source is a URI with http or https protocol. One may then argue that "well if it came from the file system wouldn't that be so fast that it could be done on the UI thread". One may argue that, but that is not the whole argument. Once the file has been read from disk it still must be decoded and then resized. While this generally is fast and would normally take less than 1 second, we should remember back to some magic number I made up that was 250ms-330ms (1/4 to 1/3 of a second) is the max the UI thread can hang for before it is very clear to the user. Now if I load just 8 images that each take 330ms-500ms to load from disk, decode and then resize this will create an awful user experience for the ender user as the progress bar stutters and images appear and the mouse is jerky.

To reproduce the problems discussed in this post you will want to get some hi-res images and load up the example code that can be found here. Try out the 6th example and point the source folder of the pop up widow to your folder with the hi-res images.

Next post we will look at the issues we will face with solving the problem of Images that kill the responsiveness of our UI. We will discuss fun things like decoders, freezables and maybe the parallel extensions that will become part of .NET 4.0

Previous - Responsive WPF User Interfaces Part 5 - Testing multi-threaded WPF code

Next - Responsive WPF User Interfaces Part 7 - Responsive Controls

Back to series Table Of Contents

Working version of the code can be found here

VS 2010 Beta

2009-05-21T09:32:00.001+01:00

Yahoo! Shiny new toys to play with http://blogs.msdn.com/wpfsdk/archive/2009/05/20/visual-studio-10-beta-1-bits-are-now-publically-available.aspx

Responsive WPF User Interfaces Part 5

2009-02-16T22:46:00.002Z

Testing multi-threaded WPF code

Stop press: Dont use this method. Read the article, but follow comment at bottom of post (ie Use Rx instead)

In the last post in this series we took a rather long way to get to the multi-threaded result we were looking for. This was because we decided to separate out our code in cohesive parts that allowed for more maintainable code. The popular method of creating maintainable code currently is through Unit Testing. An even better approach would be to apply Test Driven Development. TDD is a style of design where tests outlining our goals are written first and then code to satisfy those tests is created. Having identified the technical challenges of keeping a WPF application responsive, namely; the concepts related to the Dispatcher and parallel programming, we can now step back again to see the bigger picture as we did in part 4, to move forward.

In this post we will drive out our presentation model through tests and discover the issues related to testing WPF code and in particular testing the multi-threaded code. Just to set expectations, there is an assumption here that the reader is familiar with WPF Data Templates and unit testing with a mocking framework.

From my experience with unit testing it is always good to start off with a set of requirements that we could turn into tests. Continuing with our example of a thumbnail viewer, lets create a list of our requirements:

User can choose a folder
Can see what folder was selected
Will list all images in that directory and sub directories
Will indicate when it is processing
Cannot set folder while processing
The user interface will remain responsive during processing

For brevity, I will try to cheat as much as I can so more time can be spent on the technical issues of testing multi-threaded WPF code and less time spouting TDD goodness. I will first cheat by making the assumption that this is WPF code so will need implement constructs such as ICommand and INotifyPropertyChanged.

From these requirements I eventually came to the following test cases:

[TestClass]
class PhotoAlbumModelFixture
{
    [TestMethod]
    public void SetSourceCommand_sets_SourcePath(){}
    
    [TestMethod]
    public void SetSourceCommand_calls_FileService_with_value_from_FolderSelector(){}

    [TestMethod] 
    public void SetSourceCommand_calls_FileService_with_an_ImageOnly_filter() {}

    [TestMethod]
    public void SetSourceCommand_populates_images_from_FileService(){}

    [TestMethod]
    public void SetSourceCommand_recursively_calls_FileService(){}

    [TestMethod]
    public void SetSourceCommand_does_not_call_FileService_with_empty_value(){}

    [TestMethod]
    public void SetSourceCommand_sets_IsLoading_to_True(){}

    [TestMethod]
    public void SetSourceCommand_is_disabled_when_IsLoading(){}

    [TestMethod]
    public void FileService_is_called_Async(){}
}

When writing some of the tests I found that standard unit test code was fine. When I started to implement my code that had to make multithreaded calls, however, testing became problematic.

For example, my first attempt to write the "SetSourceCommand_calls_FileService_with_value_from_FolderSelector" test looked like this:

[TestMethod]
public void SetSourceCommand_calls_FileService_with_value_from_FolderSelector()
{
    Rhino.Mocks.MockRepository mockery = new Rhino.Mocks.MockRepository();
    ArtemisWest.Demo.ResponsiveUI._4_MultiThreaded.IFileService fileService = mockery.DynamicMock<ArtemisWest.Demo.ResponsiveUI._4_MultiThreaded.IFileService>();
    ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector folderSelector = mockery.DynamicMock<ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector>();

    PhotoAlbumModel model = new PhotoAlbumModel(Dispatcher.CurrentDispatcher, fileService, folderSelector);

    using (mockery.Record())
    {
        Expect.Call(folderSelector.SelectFolder()).Return(FolderPath);
        Expect.Call(fileService.ListFiles(FolderPath, null)).Return(EmptyList).IgnoreArguments().Constraints(
          Rhino.Mocks.Constraints.Is.Equal(FolderPath),
          Rhino.Mocks.Constraints.Is.Anything());
        Expect.Call(fileService.ListDirectories(FolderPath)).Return(EmptyList);
    }

    using (mockery.Playback())
    {
        model.SetSourcePathCommand.Execute(null);
    }
}

And following TDD rules of just write enough code to satisfy the test, my implementation of the Execute command handler looks like this:

void ExecuteSetSourcePath(object sender, ExecutedEventArgs e)
{
    SourcePath = this.folderSelector.SelectFolder();
    IsLoading = true;
    LoadPath(SourcePath);
    IsLoading = false;
}

void LoadPath(string path)
{
    IEnumerable<string> files = fileService.ListFiles(path, IsImage);
    foreach (string item in files)
    {
        Images.Add(item);
    }

    IEnumerable<string> directories = fileService.ListDirectories(path);
    foreach (string item in directories)
    {
        LoadPath(item);
    }
}

Astute readers will notice the complete lack of "Responsive UI Code". Well I have yet to reach my test that specifies that the call to FileService should be asynchronous. So after I have finished writing all of my other tests and the code to support them, I write some test code to enforce the responsive requirement:

[TestMethod]
public void FileService_is_called_Async()
{
    SlowFileServiceFake fileService = new SlowFileServiceFake();
    Rhino.Mocks.MockRepository mockery = new Rhino.Mocks.MockRepository();
    ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector folderSelector = mockery.DynamicMock<ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector>();

    PhotoAlbumModel model = new PhotoAlbumModel(Dispatcher.CurrentDispatcher, fileService, folderSelector);

    using (mockery.Record())
    {
        Expect.Call(folderSelector.SelectFolder()).Return(FolderPath);
    }

    using (mockery.Playback())
    {
        model.SetSourcePathCommand.Execute(null);
        //Due to the sleeps in the stub this should only have 1 item at this stage.
        Assert.IsTrue(model.Images.Count < fileService.ExpectedFiles.Count);
    }
    Thread.Sleep(300);
    CollectionAssert.AreEquivalent(fileService.ExpectedFiles, model.Images);
}

private class SlowFileServiceFake : ArtemisWest.Demo.ResponsiveUI._4_MultiThreaded.IFileService
{
    public readonly List<string> ExpectedFiles = new List<string>() { "value0", "value1", "value2", "value3", "value4", "value5" };

    public IEnumerable<string> ListFiles(string path, Predicate<string> filter)
    {
        foreach (var item in ExpectedFiles)
        {
            yield return item;
            Thread.Sleep(30);
        }
    }

    public IEnumerable<string> ListDirectories(string path)
    {
        return EmptyList;
    }
}

Here I have used a Fake to provide canned results with a hard coded delay. In theory this should drip feed values to the presentation model.

I now go back to update my model to get the tests to pass. Following my own advice I change the code to look like the following:

void ExecuteSetSourcePath(object sender, ExecutedEventArgs e)
{
    string folder = this.folderSelector.SelectFolder();
    if (!string.IsNullOrEmpty(folder))
    {
        SourcePath = folder;
        StartLoadingFiles(SourcePath);
    }
}

void StartLoadingFiles(string path)
{
    BackgroundWorker fileWorker = new BackgroundWorker();
    fileWorker.DoWork += (sender, e) =>
    {
        LoadPath(path);
    };
    fileWorker.RunWorkerCompleted += (sender, e) =>
    {
        IsLoading = false;
    };
    fileWorker.RunWorkerAsync();
}

void LoadPath(string path)
{
    IEnumerable<string> files = fileService.ListFiles(path, IsImage);
    foreach (string item in files)
    {
        dispatcher.Invoke(new Action(() => { Images.Add(item); }));
    }

    IEnumerable<string> directories = fileService.ListDirectories(path);
    foreach (string item in directories)
    {
        LoadPath(item);
    }
}

To my surprise this code does not satisfy the test! The reason is because the Dispatcher is not in an Executing loop. This would normally be started by WPF when the application starts, but here we are just in test code. The solution lies with the DispatcherFrame. A dispatcher frame represents an execution loop in a Dispatcher. We can kick start this loop by simply instantiating a new DispatcherFrame and calling Dispatcher.PushFrame with it as the parameter. See Dan Crevier's blog for a really simple way to unit test in this fashion. If we implement Dan's method we end up with the following test code:

[TestMethod]
public void FileService_is_called_Async()
{
    SlowFileServiceFake fileService = new SlowFileServiceFake();
    Rhino.Mocks.MockRepository mockery = new Rhino.Mocks.MockRepository();
    ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector folderSelector = mockery.DynamicMock<ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector>();

    PhotoAlbumModel model = new PhotoAlbumModel(Dispatcher.CurrentDispatcher, fileService, folderSelector);

    using (mockery.Record())
    {
        Expect.Call(folderSelector.SelectFolder()).Return(FolderPath);
    }

    DispatcherFrame frame = new DispatcherFrame();
    model.PropertyChanged += (sender, e) => 
    {
        if (e.PropertyName == "IsLoading" && !model.IsLoading)
        {
            frame.Continue = false;
        }
    };

    using (mockery.Playback())
    {
        model.SetSourcePathCommand.Execute(null);
        //Due to the sleeps in the stub this should only have 1 item at this stage.
        Assert.IsTrue(model.Images.Count < fileService.ExpectedFiles.Count);
        Dispatcher.PushFrame(frame);
    }
    Thread.Sleep(300);
    CollectionAssert.AreEquivalent(fileService.ExpectedFiles, model.Images);
}

We could call it a day here, however I am not happy enough with this style of coding. My first problem is the explicit sleep I have in there. This makes my test slow. The fastest this test will ever run is 300ms. I could replace that with some sort of loop to check progress but then my test code is looking less like test code and more like low level code to dance-around dispatchers. This raises my other issue; the clutter I have added by adding the DispatcherFrame stuff.

My solution here was to create a construct that allowed me to code my intentions in a way that I thought to be clearer. I wanted to be able to

test code that would involve multi-threaded code and the dispatcher
specify what condition defined its completion
specify a timeout
treat the code block as a blocking call

The result I came up with is the following:

[TestMethod]
public void FileService_is_called_Async()
{
    Rhino.Mocks.MockRepository mockery = new Rhino.Mocks.MockRepository();
    SlowFileServiceFake fileService = new SlowFileServiceFake();
    ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector folderSelector = mockery.DynamicMock<ArtemisWest.Demo.ResponsiveUI._5_MultiThreadedTest.IFolderSelector>();

    PhotoAlbumModel model = new PhotoAlbumModel(Dispatcher.CurrentDispatcher, fileService, folderSelector);
    using (mockery.Record())
    {
        Expect.Call(folderSelector.SelectFolder()).Return(FolderPath);
    }

    using (DispatcherTester dispatcherTester = new DispatcherTester(Dispatcher.CurrentDispatcher))
    {
        model.PropertyChanged += (sender, e) =>
        {
            if (e.PropertyName == "IsLoading" && !model.IsLoading)
                dispatcherTester.Complete();
        };
        dispatcherTester.Execute(() =>
          {
              model.SetSourcePathCommand.Execute(null);
              Assert.IsTrue(model.Images.Count < fileService.ExpectedFiles.Count);
          },
          new TimeSpan(0, 0, 2));
    }
    CollectionAssert.AreEquivalent(fileService.ExpectedFiles, model.Images);
}

This code describes a block of code that may require the use of the dispatcher, the condition where the asynchronous code is considered complete and a TimeSpan for a timeout. The test code is still not perfect but I think it is a step in the right direction.

The code for the DispatcherTester looks something like this:

internal sealed class DispatcherTester : IDisposable
{
  private readonly Dispatcher dispatcher;
  private readonly DispatcherFrame dispatcherFrame = new DispatcherFrame();

  public DispatcherTester(Dispatcher dispatcher)
  {
    this.dispatcher = dispatcher;
  }

  public Dispatcher Dispatcher
  {
    get { return dispatcher; }
  }

  public void Execute(Action action, TimeSpan timeout)
  {
    Execute(action, timeout, new TimeSpan(10));
  }

  public void Execute(Action action, TimeSpan timeout, TimeSpan wait)
  {
    Stopwatch stopwatch = Stopwatch.StartNew();
    action.Invoke();
    Dispatcher.PushFrame(dispatcherFrame);

    while (dispatcherFrame.Continue && stopwatch.Elapsed < timeout)
    {
      Thread.Sleep(wait);
    }
    if (stopwatch.Elapsed >= timeout)
    {
      dispatcherFrame.Continue = false;
      Dispatcher.DisableProcessing();
      Dispatcher.ExitAllFrames();
      throw new TimeoutException();
    }
  }

  public void Complete()
  {
    dispatcherFrame.Continue = false;
  }

  #region IDisposable Members

  public void Dispose()
  {
    dispatcherFrame.Continue = false;
  }

  #endregion
}

If any other developers have hit problems trying to write unit tests for their WPF code then I hope that this snippet of code helps. For those that are interested, the end result for the Model looks like

public class PhotoAlbumModel : INotifyPropertyChanged
{
  #region Fields
  private readonly Dispatcher dispatcher;
  private readonly _4_MultiThreaded.IFileService fileService;
  private readonly IFolderSelector folderSelector;
  private readonly DelegateCommand setSourcePathCommand;
  private readonly ObservableCollection<string> images = new ObservableCollection<string>();
  private string sourcePath;
  private bool isLoading = false;
  #endregion

  public PhotoAlbumModel(Dispatcher dispatcher, _4_MultiThreaded.IFileService fileService, IFolderSelector folderSelector)
  {
    this.dispatcher = dispatcher;
    this.fileService = fileService;
    this.folderSelector = folderSelector;
    setSourcePathCommand = new DelegateCommand(ExecuteSetSourcePath, CanSetSourcePath);
    this.PropertyChanged += (sender, e) => { if (e.PropertyName == "IsLoading") setSourcePathCommand.OnCanExecuteChanged(); };
  }

  public ObservableCollection<string> Images
  {
    get { return images; }
  }

  public bool IsLoading
  {
    get { return isLoading; }
    private set
    {
      isLoading = value;
      OnPropertyChanged("IsLoading");
    }
  }

  public string SourcePath
  {
    get { return sourcePath; }
    private set
    {
      sourcePath = value;
      OnPropertyChanged("SourcePath");
    }
  }

  public ICommand SetSourcePathCommand { get { return setSourcePathCommand; } }

  #region Command handlers
  void CanSetSourcePath(object sender, CanExecuteEventArgs e)
  {
    e.CanExecute = !IsLoading;
  }

  void ExecuteSetSourcePath(object sender, ExecutedEventArgs e)
  {
    string folder = this.folderSelector.SelectFolder();
    if (!string.IsNullOrEmpty(folder))
    {
      SourcePath = folder;
      StartLoadingFiles(SourcePath);
    }
  }
  #endregion

  #region Private method
  void StartLoadingFiles(string path)
  {
    IsLoading = true;
    BackgroundWorker fileWorker = new BackgroundWorker();
    fileWorker.DoWork += (sender, e) =>
    {
      LoadPath(path);
    };
    fileWorker.RunWorkerCompleted += (sender, e) =>
    {
      IsLoading = false;
    };
    fileWorker.RunWorkerAsync();
  }

  void LoadPath(string path)
  {
    IEnumerable<string> files = fileService.ListFiles(path, IsImage);
    foreach (string item in files)
    {
      dispatcher.Invoke(new Action(() => { Images.Add(item); }));
    }

    IEnumerable<string> directories = fileService.ListDirectories(path);
    foreach (string item in directories)
    {
      LoadPath(item);
    }
  }

  bool IsImage(string file)
  {
    string extension = file.ToLower().Substring(file.Length - 4);
    switch (extension)
    {
      case ".bmp":
      case ".gif":
      case ".jpg":
      case ".png":
        return true;
      default:
        return false;
    }
  }
  #endregion

  #region INotifyPropertyChanged Members
  public event PropertyChangedEventHandler PropertyChanged;

  protected void OnPropertyChanged(string propertyName)
  {
    PropertyChangedEventHandler handler = PropertyChanged;
    if (handler != null)
    {
      handler(this, new PropertyChangedEventArgs(propertyName));
    }
  }
  #endregion
}

Notice that the Folder dialogue code from the previous post has now been pushed out to an interface which cleans up the code and also allows us to test this model.

The data template looks almost identical to the previous post's window xaml:

<DataTemplate DataType="{x:Type local:PhotoAlbumModel}">
  <DataTemplate.Resources>
    <BooleanToVisibilityConverter x:Key="boolToVisConverter"/>
  </DataTemplate.Resources>
  <DockPanel>
    <DockPanel DockPanel.Dock="Top">
      <ProgressBar IsIndeterminate="True" Height="18" 
                   DockPanel.Dock="Top" 
                   Visibility="{Binding Path=IsLoading, 
                                Converter={StaticResource boolToVisConverter}}" />
      <Button x:Name="SetSourcePath" Command="{Binding Path=SetSourcePathCommand}" 
              DockPanel.Dock="Right">Set Source</Button>
      <Border BorderThickness="1" BorderBrush="LightBlue" Margin="3">
        <Grid>
          <TextBlock Text="{Binding SourcePath}">
            <TextBlock.Style>
              <Style TargetType="TextBlock">
                <Setter Property="Visibility" Value="Visible"/>
                <Style.Triggers>
                  <DataTrigger Binding="{Binding SourcePath}" Value="">
                    <Setter Property="Visibility" Value="Collapsed"/>
                    </DataTrigger>
                  <DataTrigger Binding="{Binding SourcePath}" Value="{x:Null}">
                    <Setter Property="Visibility" Value="Collapsed"/>
                  </DataTrigger>
                </Style.Triggers>
              </Style>
            </TextBlock.Style>
          </TextBlock>
          <TextBlock Text="Source not set" Foreground="LightGray" FontStyle="Italic">
            <TextBlock.Style>
              <Style TargetType="TextBlock">
                <Setter Property="Visibility" Value="Collapsed"/>
                <Style.Triggers>
                  <DataTrigger Binding="{Binding SourcePath}" Value="">
                    <Setter Property="Visibility" Value="Visible"/>
                    </DataTrigger>
                  <DataTrigger Binding="{Binding SourcePath}" Value="{x:Null}">
                    <Setter Property="Visibility" Value="Visible"/>
                  </DataTrigger>
                </Style.Triggers>
              </Style>
            </TextBlock.Style>
          </TextBlock>
        </Grid>
      </Border>
    </DockPanel>
    <ListView ItemsSource="{Binding Images}" 
              ScrollViewer.VerticalScrollBarVisibility="Visible" />
  </DockPanel>
</DataTemplate>

I hope that the examples here

give some insight on how you too can unit test your code,
show why a model is a better option for your WPF applications as you can test them
give you the courage to write multi threaded code in your WPF models in the knowledge that you can test it

Next we discover that sometimes controls can cause an unresponsive UI in Part 6.

Previous - Responsive WPF User Interfaces Part 4 - Multi-threaded Programming in WPF.

Back to series Table Of Contents

Working version of the code can be found here