usa RX per attivare eventi in momenti diversi?

Question

Ho una vasta collezione di classe semplice coppia:

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

Essi sono ordinati per il Timestamp ascendente. Voglio attivare un evento con il valore (ad esempio, azione <doppio>) per ciascun elemento nell'elenco al momento opportuno. I tempi sono passati, quindi ho bisogno di normalizzare i timestamp in modo tale che il primo della lista sia "now". Possiamo impostarlo con le Reactive Extensions in modo tale da attivare l'evento successivo dopo la differenza di tempo tra due elementi?

Answer 1

Say pairs è la sequenza:

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

Ora obs è coppie come un'osservabile ordinato.

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

Lo zip si occupa di trasformare i tempi assoluti in offset. Che prenderà la sequenza e unirsi con se stesso, ma compensato da uno, come segue

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

Questo nuovo valore viene poi messo in Observable.Timer ritardarla la quantità appropriata. L'ultimo Concat appiattisce il risultato di un IObservable<IObservable<double>> in un IObservable<double>. Questo presuppone che la sequenza sia ordinata.

Answer 2

Penso che questo problema sia interessante, questo sarebbe il mio primo tentativo.

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

Se per "usando Rx" mi permette di usare solo i pianificatori Rx, allora questa è una soluzione molto semplice:

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))); 

Questo utilizza la System.Interactive estensione ForEach, ma si potrebbe utilizzare un ciclo regolare foreach per caricare lo scheduler.

Ho testato il codice con i seguenti dati dummy:

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, }, 
}; 

Spero che questo aiuta.