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

Question

在一个IObservable序列中(在.NET的Reactive Extensions中),我想获取previous和current元素的值,以便我可以比较它们.我在网上发现了一个类似于下面的例子来完成任务:

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

Answer 2

@James World 附录对我来说看起来很棒，如果不是因为Tuple<>，我几乎总是不喜欢：“ .Item1 是前一个吗？还是当前的？我不记得了。选择器的第一个参数是什么，是吗？上一项？ ”。

对于那部分，我喜欢@dcstraw 专用的ItemWithPrevious<T>. 因此，您可以将两者放在一起（希望我没有将以前的与当前的混淆）并进行一些重命名和设施：

public static class ObservableExtensions
{
    public static IObservable<SortedPair<TSource>> CombineWithPrevious<TSource>(
        this IObservable<TSource> source, 
        TSource initialValue = default(TSource))
    {
        var seed = SortedPair.Create(initialValue, initialValue);

        return source.Scan(seed,
            (acc, current) => SortedPair.Create(current, acc.Current));
    }

    public static IObservable<TResult> CombineWithPrevious<TSource, TResult>(
        this IObservable<TSource> source,
        Func<SortedPair<TSource>, TResult> resultSelector,
        TSource initialValue = default(TSource))
    {
        var seed = SortedPair.Create(initialValue, initialValue);

        return source
            .Scan(seed,
                (acc, current) => SortedPair.Create(current, acc.Current))
            .Select(p => resultSelector(p));
    }
}

public class SortedPair<T>
{
    public SortedPair(T current, T previous)
    {
        Current = current;
        Previous = previous;
    }

    public SortedPair(T current) : this(current, default(T)) { }

    public SortedPair() : this(default(T), default(T)) { }

    public T Current;
    public T Previous;
}

public class SortedPair
{
    public static SortedPair<T> Create<T>(T current, T previous)
    {
        return new SortedPair<T>(current, previous);
    }

    public static SortedPair<T> Create<T>(T current)
    {
        return new SortedPair<T>(current);
    }

    public static SortedPair<T> Create<T>()
    {
        return new SortedPair<T>();
    }
}