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
