XAML Playground
about XAML and other Amenities

Taking advantage of combining different streams using Reactive Extension's:

2011-04-04T16:18:45+01:00 by Andrea Boschin

One interesting feature of Reactive Extensions is the combining of Observables. There are a number of extension methods made to combine two or more streams; Amb, Concat, SelectMany, Merge, Zip, CombineLatest, all these methods are made to take, multiple and non necessarily omogeneous, streams and combine them in a unique resulting stream based on different rules. Here a brief resume of the rules applied from every method

Amb returns the stream that start providing values by first
Concat chain two streams one to the end of the other and create a single output
SelectMany Returns every value from the second stream for each value from the first
Merge returns the two streams merged basing of when each source returns its value
Zip returns values from one stream paired with values from another, only when a couple is available.
CombineLatest Combine two streams returning always the latest from both, for each occurrence of a value.


Understanding the meaning of each method is really difficult if you do not make some simple experiment, but this is out of the scope of this post. Now I would like to show the meaning of "combining two streams" with a real example that show how a little change in the previous mouse trail example can give a great difference in terms of functions.

In my last example I used the TrailWithCount and TrailWithTime methods to catch MouseMove events to draw a polyline on the plugin that il the trail of the mouse pointer. In the example you have seen the trail shortens while the mouse was moving but when the mouse stops also the trail stops. Now I want to change the example to let the trail shorten to zero length when the mouse stops moving.

To achieve this result I need to have a timer that continues to record the current position for the mouse when it stops. For this purpose I can use the static method Obsevable.Interval(timespan) that is able to generate a stream of events at the given interval of time. Here is the two lines I have to add to the previous example.

   1: Observable.FromEvent<MouseEventHandler, MouseEventArgs>(
   2:     ev => new MouseEventHandler(ev),
   3:     ev => this.LayoutRoot.MouseMove += ev,
   4:     ev => this.LayoutRoot.MouseMove -= ev)
   5:     .CombineLatest(Observable.Interval(TimeSpan.FromMilliseconds(25)), (a, b) => a)
   6:     .ObserveOnDispatcher()
   7:     .Select(o => o.EventArgs.GetPosition(this.LayoutRoot))
   8:     .TrailWithCount(100)
   9:     .Subscribe(DrawLine);

the CombineLatest method gets events from the mouse and from the time interval (this regularly every 25 milliseconds) and using the selector lambda it takes from the combined stream only the mouse events. Every time an interval expire the combined value will contain an integer (the number of intervals from the start) and the latest mouse event. When the mouse stops moving the stream from the MouseMove also stops to provide values but the CombineLatest method will replicate the latest value taken for every interval elapsed. So the trail will be feeded with the latest position and the visual effect will be the shorten of the trail until it reaches the current position of the mouse pointer.

The ObserveOnDispatched method is required because the Observable.Interval method implies the use of a separate thread so when we get the values we are in this thread and we need to marshall back to the UI thread. This method does the trick. Here is the result:

Get Microsoft Silverlight