usar RX para desencadenar eventos en diferentes momentos?

Question

Tengo una gran colección de clase simple par:

public class Pair { public DateTime Timestamp; public double Value; } 

Están ordenados por la marca de tiempo ascendente. Quiero activar un evento con el valor (por ejemplo, acción <doble>) para cada elemento en la lista en el momento apropiado. Los tiempos están en el pasado, así que necesito normalizar las marcas de tiempo de modo que el primero en la lista sea "ahora". ¿Podemos configurar esto con las Extensiones Reactivas de modo que desencadene el siguiente evento después de la diferencia de tiempo entre dos elementos?

Answer 1

Say pairs es su secuencia:

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

Ahora es obs pares como un observable ordenada.

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

El zip se encarga de convertir los tiempos absolutos en desplazamientos. Se llevará a la secuencia y unirlo con ella misma, pero compensado por una, de la siguiente manera

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

Este nuevo valor se pone entonces en el Observable.Timer retrasarla la cantidad apropiada. El Concat final aplana el resultado de IObservable<IObservable<double>> en un IObservable<double>. Esto supone que tu secuencia está ordenada.

Answer 2

Creo que este problema es interesante, este sería mi primer intento.

static void RunPairs(IEnumerable<Pair> pairs, Action<double> pairEvent) 
{ 
    if (pairs == null || !pairs.Any() || pairEvent == null) 
    return; 

    // if we can promise the pairs are already sorted 
    // obviously we don't need this next line 
    pairs = pairs.OrderBy(p => p.Timestamp); 
    var first = pairs .First().Timestamp; 
    var wrapped = pairs.Select(p => new { Offset = (p.Timestamp - first), Pair = p }); 

    var start = DateTime.Now; 

    double interval = 250; // 1/4 second 
    Timer timer = new Timer(interval); 

    timer.AutoReset = true; 
    timer.Elapsed += (sender, elapsedArgs) => 
    { 
    var signalTime = elapsedArgs.SignalTime; 
    var elapsedTime = (signalTime - start); 

    var pairsToTrigger = wrapped.TakeWhile(wrap => elapsedTime > wrap.Offset).Select(w => w.Pair); 
    wrapped = wrapped.Skip(pairsToTrigger.Count()); 

    if (!wrapped.Any()) 
     timer.Stop(); 

    foreach (var pair in pairsToTrigger) 
     pairEvent(pair.Value);  
    }; 

    timer.Start(); 
} 

Answer 3

Si por "el uso de Rx" me permiten sólo tiene que utilizar los programadores de Rx, entonces esta es una solución muy fácil:

Action<double> action = 
    x => 
     Console.WriteLine(x); 

var ts0 = pairs.Select(p => p.Timestamp).Min(); 

pairs 
    .ForEach(p => 
     Scheduler 
      .ThreadPool 
      .Schedule(
       p.Timestamp.Subtract(ts0), 
       () => action(p.Value))); 

Este utiliza el System.Interactive extensión ForEach, pero sólo podía utilizar un bucle regular foreach para cargar el programador.

He probado el código con los siguientes datos ficticios:

var pairs = new [] 
{ 
    new Pair { Timestamp = new DateTime(2011, 1, 1, 7, 12, 30), Value = 1.1, }, 
    new Pair { Timestamp = new DateTime(2011, 1, 1, 7, 12, 45), Value = 1.2, }, 
    new Pair { Timestamp = new DateTime(2011, 1, 1, 7, 12, 40), Value = 1.3, }, 
}; 

espero que esto ayude.