Andrej Tozon's blog

In my last blog post I wrote about using Reactive Extensions together with layout states in Silverlight to gradually introduce collections of data to a ListBoxes. One small problem with code from that post is that I’m generating the sequence of data myself, whereas in real-world scenario would be pulling it from some data source like database or web service. It’s easy to turn an existing collection to an Observable by using the ToObservable() operator, but the generated sequence wouldn’t be time-based, as it would have been if we had used the GenerateWithTime() constructor.

So how do we turn an existing enumerable collection into a time-based observable sequence?

First, we need to convert the enumerable to a plain observable. Here’s the example of such conversion:

private readonly string text = "REACTIVE";

IObservable<char> letters = text.ToObservable();

[Reminder: a string is a collection/sequence of chars – turning this collection into an observable sequence results in IObservable<char>]

Next, we need to create an observable sequence that will serve as a “beat” – a time-based sequence that would define points in time at which we want to “release” (or trigger) the next item in our data collection. The following code will create a fast “beat”, “thumping” at every 300 ms:

IObservable<long> beat = Observable.Interval(TimeSpan.FromSeconds(.3));

Now we only need to lay the numbers from the enumerable collection over created beat and for that, there is a convenient combinator operator available in Reactive Extensions. It’s called – Zip.

IObservable<string> dancingLetters = numbers.Zip(beat, (letter, time) => letter);

Zip operator will take two observable sequences (letters and beat) and produce a new value pair when a new value is present in both sequences, kind of like a zipper. In our case, the letters sequence starts with 8 values and beat starts with zero. As the beat sequence starts producing new values (one every 0.3 seconds), these new values are paired (zipped) with values from letters, resulting in a new pair of data (letter, time) releasing every 0.3 seconds. As we’re only interested in numbers from the numbers sequence, we return only those values, ignoring values values from the beat sequence (=> letter).

dancingLetters is now a time-based observable sequence which we can subscribe to:

dancingLetters.ObserveOnDispatcher().Subscribe(AddLetter);

… and get the same result as in previous post, only with letters.

One great thing about Reactive Extensions is they are very extensible, meaning you can write your own operators by creating new extension methods. For frequent operations like the one we’ve just performed, it’s a perfect fit. Here’s an example of an OnTimeline() operator that would put an observable sequence on a live timeline:

public static class ObservableEx
{
    public static IObservable<TSource> OnTimeline<TSource>(this IObservable<TSource> source, TimeSpan period)
    {
        return source.Zip(Observable.Interval(period), (d, t) => d);
    }
}

With the new operator handy and in place, we could rewrite the previous snippet as (full code ahead):

private readonly string text = "REACTIVE";

private void OnLoaded(object sender, RoutedEventArgs e)
{
    text.ToObservable()
        .OnTimeline(TimeSpan.FromSeconds(.3))
        .ObserveOnDispatcher()
        .Subscribe(AddLetter);
}

private void AddLetter(char letter)
{
    list.Items.Add(letter);
}

This application in action can be observed through this link

In this post, I extended the code sample from previous post by creating a new extension method that puts an existing observable sequence on a timeline, with delayed item notification.

Download source code from here.