use RX to trigger events at varying times?

c#, multithreading, system.reactive

Solution

Say `pairs` is your sequence:

var obs = pairs.OrderBy(p => p.Timestamp).ToObservable();

Now `obs` is pairs as an ordered observable.

Observable.Zip(
    obs,
    obs.Take(1).Concat(obs),
    (pair1, pair2) => Observable.Timer(pair1.Timestamp - pair2.Timestamp)
      .Select(_ => pair1.Value))
.Concat()
.Subscribe(/* Do something here */);

The zip takes care of turning the absolute times into offsets. It will take the sequence and join it with itself, but offset by one, as follows

Original 1--2--4--7--11
Offset   1--1--2--4--7--11
Joined   0--1--2--3--4

This new value is then put into the `Observable.Timer` to delay it the appropriate amount. The final `Concat` flattens the result from an `IObservable<IObservable<double>>` into an `IObservable<double>`. This assumes that your sequence is ordered.

Problem

I've got a large collection of simple pair class: ``` public class Pair { public DateTime Timestamp; public double Value; } ``` They are sorted by the ascending Timestamp. I want to trigger an event with the Value (say, Action<double>) for each item in the list at the appropriate time. The times are in the past, so I need to normalize the Timestamps such that the first one in the list is "now". Can we set this up with the Reactive Extensions such that it triggers the next event after on the difference in time between two items?

Original source