RX Frequency Cheat Sheet

Ok, so you’re down with observables and you get the use of where and select statements to restrict and project your observables/events.  Now you want to filter on a time frequency.

Here’s a quick way to visually plan out your pipeline with RX.

Basic Subscribe

The basics, you want to know about every single event:

var observable = Enumerable.Range(1, 1000000).ToObservable();
using (observable.Subscribe(Console.WriteLine))

Sample

You want to get the latest event every X seconds.

var observable = Enumerable.Range(1, 1000000).ToObservable();
using (observable
       .Sample(TimeSpan.FromSeconds(1))
       .Subscribe(Console.WriteLine))

Regulate (custom extension method)

Slow down the events to a manageable speed without losing any events.

Use the extension method provided by John Rayner

http://sharpfellows.com/post/Rx-Controlling-frequency-of-events.aspx

var observable = Enumerable.Range(1, 1000000).ToObservable();
using (observable
       .Regulate(TimeSpan.FromSeconds(1))
       .Subscribe(Console.WriteLine))

Buffer

Receive all events every X seconds.

IObservable<long> observable = Observable.Interval(TimeSpan.FromMilliseconds(250));
using (observable
        .Buffer(TimeSpan.FromSeconds(1))
        .Subscribe(x => { foreach (var item in x) Console.WriteLine(item); }))
{
    Console.WriteLine("Press any key to unsubscribe");
    Console.ReadKey();
}

 

Here’s a few other useful tricks:

Merge

Merge two event streams into one.

IObservable<int> observable1 = (new[] { 1, 3, 5, 7, 9 }).ToObservable();
IObservable<int> observable2 = (new[] { 2, 4, 6, 8, 10 }).ToObservable();

using (observable1
       .Merge(observable2)
       .Subscribe(Console.WriteLine))

Or like this:

ConCat

Concatenates two streams.

IObservable<int> observable1 = (new[] { 1, 3, 5, 7, 9 }).ToObservable();
IObservable<int> observable2 = (new[] { 2, 4, 6, 8, 10 }).ToObservable();

using (observable1
       .Concat(observable2)
       .Subscribe(Console.WriteLine))

Zip

From two streams, takes an event from each, applies a selector logic and returns an output.

IObservable<int> observable1 = (new[] {1, 3, 5, 7, 9}).ToObservable();
IObservable<int> observable2 = (new[] {2, 4, 6, 8, 10}).ToObservable();

using (observable1
    .Zip(observable2, (x, y) => x > y ? x : y)
    .Subscribe(Console.WriteLine))

Distinct Until Changed

Only receive events that are different from the previous event.

IObservable<int> observable = (new[] { 1, 1, 1, 2, 2, 3, 4, 4 })
       .ToObservable();
using (observable
       .DistinctUntilChanged()
       .Subscribe(Console.WriteLine))

Interval

Fires the event every specified duration.

IObservable<long> observable = Observable.Interval(TimeSpan.FromSeconds(1));

using (observable.Subscribe(Console.WriteLine))
{
    Console.WriteLine("Press any key to unsubscribe");
    Console.ReadKey();
}

Delay

Delays the start of the event stream, preserving the interval between events.

IObservable<long> observable = Observable.Interval(TimeSpan.FromSeconds(0.5));

using (observable
    .Delay(TimeSpan.FromSeconds(1))
    .Subscribe(Console.WriteLine))
{
    Console.WriteLine("Press any key to unsubscribe");
    Console.ReadKey();
}

TimeInterval

Wraps the event, exposing the event as it’s value and an Interval property.  The interval is the amount of time since the previous event.

var observable = Observable.Interval(TimeSpan.FromMilliseconds(750)).TimeInterval();

using (observable.Subscribe(
    x => Console.WriteLine("{0}: {1}", x.Value, x.Interval)))
{
    Console.WriteLine("Press any key to unsubscribe");
    Console.ReadKey();
}

Timestamp

Wraps the event, exposing the event as it’s value and a timestamp property.

var observable = Observable.Interval(TimeSpan.FromSeconds(1)).Timestamp();

using (observable.Subscribe(
    x => Console.WriteLine("{0}: {1}", x.Value, x.Timestamp)))
{
    Console.WriteLine("Press any key to unsubscribe");
    Console.ReadKey();
}

Throttle

Holds back on receiving events until the throttle duration has elapsed between two events.

var dictionarySuggest = userInput
                        .Throttle(TimeSpan.FromMilliseconds(250))
                        .SelectMany(input => serverCall(input));