如何将可观察序列"延迟"一个值？

Question

在响应式扩展中，如何将可观察序列"延迟"一个值？例如：

original: 2 3 5 7 9   
delayed:    2 3 5 7

为了清楚起见，我想将序列延迟一步。这与延迟恒定时间不同。

Answer 1

试试这个：

var delayedByOne = source.Zip(source.Skip(1), (x, _) => x);

但是，如果您有冷可观察源，则可能需要执行此操作以避免运行两个并发源：

var delayedByOne = source.Publish(s => s.Zip(s.Skip(1), (x, _) => x));

Answer 2

不需要发布，但它需要采用一种方法来获得良好的类型推断，如下所示：

public static class ObservableExtensions
{
    public static IObservable<T> DelayByOne<T>(
        this IObservable<T> source)
    {
        return source.Scan(
            Tuple.Create(default(T), default(T)),
                (a, i) => Tuple.Create(i,a.Item1))
            .Select(a => a.Item2).Skip(1);
    }
}

这个

想法源于这个 http://www.zerobugbuild.com/?p=213#comment-884。可以看到上面真的是一个更通用的比较功能的专业化。从评论到我的原始博客文章，以下是用于比较当前和先前事件的修订实用程序函数：

// The result selector is passed (curr, prev) and
// prev is default(T) for first event
public static class ObservableExtensions
{
    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)),
                (a, i) => Tuple.Create(i,a.Item1))
            .Select(a => resultSelector(a.Item1, a.Item2));
    }
}

所以你可以像这样使用上面的：

var delayedByOne = source.CombineWithPrevious((x,y) => y).Skip(1);

Answer 3

刚刚意识到有一种更简洁的更短的方法：

var delayedByOne = source.Publish(ps => ps.Delay(_ => ps));