反应性扩展 (Rx) - 当间隔中没有值时，具有最后一个已知值的样本 [英] Reactive Extensions (Rx) - sample with last known value when no value is present in interval

问题描述

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

I have an observable stream that produces values at inconsistent intervals like this:

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

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

And I would like to sample this but without any empty samples once the a value has been produced:

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

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

我显然认为 Replay().RefCount() 可以在此处用于为 Sample() 提供最后一个已知值，但因为它不会重新订阅到源流，它没有解决.

I obviously thought Replay().RefCount() could be used here to provide the last known value to Sample() but as it doesn't re-subscribe to the source stream it didn't work out.

对我如何做到这一点有任何想法吗?

Any thoughts on how I can do this?

推荐答案

假设你的源流是 IObservablexs 那么你的采样间隔是 Timespan duration 那么:

Assuming your source stream is IObservable<int> xs then and your sampling interval is Timespan duration then:

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

对于通用解决方案，将 0 参数替换为 MostRecent 为 default(T) 其中 IObservablecode> 是源流类型.

For a generic solution, replace the 0 parameter to MostRecent with default(T) where IObservable<T> is the source stream type.

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

The purpose of Publish is to prevent subscription side effects since we need to subscribe to the source twice - once for MostRecent and once for SkipUntil. The purpose of the latter is to prevent sampling values until the source stream's first event.

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

You can simplify this if you don't care about getting default values before the source stream's first event:

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

一个相关的操作符 WithLatestFrom 可能也很有趣；这将在下一个版本中加入 Rx.请参阅此处了解详情.

A related operator WithLatestFrom might also be of interest; this is coming to Rx in the next release. See here for details.

这篇关于反应性扩展 (Rx) - 当间隔中没有值时，具有最后一个已知值的样本的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！