将 IObservable<Timestamped<T>> 转换为 IObservable<TimeInterval<T>>

Question

如何将Timestamped<T>的可观察序列转换为TimeInterval<T>的序列，其中间隔是原始序列上的时间戳之间的时间?

给定输入序列…

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

. .输出将是:

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

我认为这样就行了。

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

我唯一不确定的是，如果Zip结束时任何一个序列结束。我认为它可以，但我还没有测试过。

Answer 2

也许你可以结合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));
}

使用:

var intervals = stampedData.ToTimeInterval();

Answer 3

我不太了解可观察对象，但你能不能:

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

当然，这是不可能高效的，特别是如果您知道日志语句是按顺序排列的。