c# – 平滑Rx Observables

非常类似于这个问题： Rx IObservable buffering to smooth out bursts of events,我有兴趣平滑可能在爆发中发生的可观察量.

希望下面的图表说明我的目标是：

Raw:       A--B--CDE-F--------------G-----------------------
Interval:  o--o--o--o--o--o--o--o--o--o--o--o--o--o--o--o--o
Output:    A--B--C--D--E--F-----------G---------------------

鉴于原始流,我希望定期延长这些事件.

限制不起作用,因此我最终失去了原始序列的元素.

如果原始流比计时器更频繁,则Zip工作良好,但如果存在没有原始事件的时段,则Zip失败.

编辑

在回应Dan’s answer时,Buffer的问题在于,如果很多事件的突发在很短的时间间隔内到达,那么我经常收到事件.下面显示了缓冲区大小为3时可能发生的情况,以及配置为所需时间间隔的超时：

Raw:       -ABC-DEF-----------G-H-------------------------------
Interval:  o--------o--------o--------o--------o--------o--------
Buffered:  ---A---D-------------------G--------------------------
              B   E                   H
              C   F
Desired:   ---------A--------B--------C--------D--------E ..etc.

解决方法

这个怎么样？ (灵感来自 James’ answer mentioned in the comments)……

public static IObservable<T> Regulate<T>(this IObservable<T> source,TimeSpan period)
{
    var interval = Observable.Interval(period).Publish().RefCount();

    return source.Select(x => Observable.Return(x)
                                        .CombineLatest(interval,(v,_) => v)
                                        .Take(1))
                 .Concat();
}

它将原始observable中的每个值转换为自己的可观察值. CombineLatest意味着在间隔之前它不会产生值.然后我们从这些可观察的每个中取一个值并连接起来.

原始可观察量中的第一个值延迟一个周期.我不确定这是否是一个问题.