c# - 在不重新评估序列的情况下获取 IObservable 中的前一个元素

Question

在 IObservable 序列中(在 Reactive Extensions for .NET 中)，我想获取前一个元素和当前元素的值，以便我可以比较它们。我在网上找到了一个类似于下面的示例来完成任务:

sequence.Zip(sequence.Skip(1), (prev, cur) => new { Previous = prev, Current = cur })

它工作正常，只是它对序列求值两次，我想避免这种情况。您可以看到使用此代码对其进行了两次评估:

var debugSequence = sequence.Do(item => Debug.WriteLine("Retrieved an element from sequence"));
debugSequence.Zip(debugSequence.Skip(1), (prev, cur) => new { Previous = prev, Current = cur }).Subscribe();

输出显示的调试行数是序列中元素数的两倍。

我明白为什么会这样，但到目前为止我还没有找到不对序列求值两次的替代方案。如何仅通过一个序列评估将之前和当前结合起来？

Answer 1

我认为有一个更好的解决方案，即使用 Observable.Scan 并避免双重订阅:

public static IObservable<Tuple<TSource, TSource>>
    PairWithPrevious<TSource>(this IObservable<TSource> source)
{
    return source.Scan(
        Tuple.Create(default(TSource), default(TSource)),
        (acc, current) => Tuple.Create(acc.Item2, current));
}

我已经把它写在我的博客上了:http://www.zerobugbuild.com/?p=213

附录

进一步的修改允许您通过使用结果选择器更干净地处理任意类型:

public static IObservable<TResult> CombineWithPrevious<TSource,TResult>(
    this IObservable<TSource> source,
    Func<TSource, TSource, TResult> resultSelector)
{
    return source.Scan(
        Tuple.Create(default(TSource), default(TSource)),
        (previous, current) => Tuple.Create(previous.Item2, current))
        .Select(t => resultSelector(t.Item1, t.Item2));
}