c# - 平滑 Rx Observable

Question

与这个问题非常相似: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 运行良好，但如果有一段时间没有原始事件，则失败。

编辑

回应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.

Answer 1

这个怎么样？ (灵感来自 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 中的每个值转换为它自己的 observable。 CombineLatest 意味着它不会产生一个值，直到间隔产生。然后我们只从这些可观察量中的每一个中取一个值并连接起来。

原始可观察对象中的第一个值被延迟了一个周期。我不确定这对您来说是否是个问题。