Convert IObservable <Timestamp <T>> to IObservable <TimeInterval <T>>

Question

How to convert the observed Timestamped<T> sequence to the TimeInterval<T> sequence, where the interval is the time between the timestamps in the original sequence?

Given the input sequence.

 new Timestamped<int>(1, DateTime.Parse("2000-01-01 00:00:01")) new Timestamped<int>(2, DateTime.Parse("2000-01-01 00:00:05")) new Timestamped<int>(3, DateTime.Parse("2000-01-01 00:01:04")) 

.. the output will be:

 new TimeInterval<int>(1, TimeSpan.Parse("00:00:00")) new TimeInterval<int>(2, TimeSpan.Parse("00:00:04")) new TimeInterval<int>(3, TimeSpan.Parse("00:00:59")) 

Answer 1

I think the way it is.

 var s = source.Publish().RefCount(); var sprev = s.Take(1).Concat(s); var scurrent = s; var converted = Observable.Zip(sprev, scurrent, (prev, current) => new TimeInterval<int>(current.Value, current.Timestamp - prev.Timestamp)); 

The only thing I'm not sure is if the Zip ends when any sequence ends. I suppose this is so, but I have not tested it.

Answer 2

Perhaps you can use a simple projection in combination with Do :

 static IObservable<TimeInterval<T>> ToTimeInterval<T>( this IObservable<Timestamped<T>> source) { DateTimeOffset? previous = null; return source.Select(ts => new { Timestamp = ts.Timestamp, Value = ts.Value, TimeSpan = previous.HasValue ? ts.Timestamp - previous : TimeSpan.FromSeconds(0) }) .Do(xx => { previous = xx.Timestamp; }) .Select(xx => new TimeInterval<T>(xx.Value, xx.TimeSpan)); } 

Used as:

 var intervals = stampedData.ToTimeInterval(); 

Answer 3

I know little about observables, but you could do:

  myInputSequence.ToEnumerable().Select(t => new TimeInterval<int>( t.Value, t.Value == 1 ? new TimeSpan(0) : t.Timestamp - System.Reactive.Linq.Observable.ToEnumerable(myList).FirstOrDefault(t2 => t2.Value == t.Value - 1).Timestamp) ).ToObservable(); 

Of course, this is not so effective, especially if you know that the log statements are in order.