c# - Reactive Extensions (Rx) - 当间隔中没有值时，使用最后一个已知值进行采样

Question

我有一个可观察流，它以不一致的间隔产生值，如下所示:

------1---2------3----------------4--------------5---

我想对此进行采样，但在生成 a 值后没有任何空样本:

------1---2------3----------------4--------------5-----

----_----1----2----3----3----3----4----4----4----5----5

我显然认为可以在此处使用 Replay().RefCount() 为 Sample() 提供最后已知的值，但因为它不会重新订阅到源流，它没有成功。

关于如何做到这一点有什么想法吗？

Answer 1

假设您的源流是 IObservable<int> xs那么你的采样间隔是Timespan duration然后:

xs.Publish(ps => 
    Observable.Interval(duration)
        .Zip(ps.MostRecent(0), (x,y) => y)
        .SkipUntil(ps))

对于通用解决方案，替换 0参数 MostRecent与 default(T)其中 IObservable<T>是源码流类型。

Publish的目的|是为了防止订阅副作用，因为我们需要订阅源两次 - 一次是 MostRecent一次是 SkipUntil .后者的目的是防止在源流的第一个事件发生之前对值进行采样。

如果您不关心在源流的第一个事件之前获取默认值，您可以简化它:

Observable.Interval(duration)
    .Zip(xs.MostRecent(0), (x,y) => y)

相关运营商WithLatestFrom可能也有兴趣；这将在下一个版本中出现在 Rx 中。参见 here了解详情。