反应性扩展:具有不同间隔的节流/采样

Question

我有一个IObservable随机生成值,我想限制这个序列.我发现的一件事是,Throttle运营商对"节流"的定义与我的不同.

Throttle只有在指定的时间间隔过去后才会产生静音值(它会产生最后看到的值).我认为节流意味着在指定的时间间隔产生价值(当然,除非有沉默).

说,我期望Observable.Interval(100).Select((_,i) => i).Throttle(200)产生(模数任何性能/时间问题)偶数,因为我把它限制为"半速".然而,这个序列根本没有产生任何价值,因为从来没有一段长度为200的沉默.

所以,我发现Sample实际上我想要的"节流"行为.Observable.Interval(100).Select((_,i) => i).Sample(200)产生(再次,模数任何性能/时间问题)偶数序列.

但是,我还有另外一个问题:间隔会有所不同,具体取决于最后的"采样"值.我想要的是编写一个如下所示的运算符:

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector);

该intervalSelector参数生成下一个样本的间隔,第一个样本...是从第一个值或另一个参数获取,我不在乎.

我试着写这篇文章,但最后我的结构很复杂,不能正常工作.我的问题是,我可以使用现有的运算符(也就是单行代码)来构建它吗？

Answer 1

很多个小时之后,有一些睡眠,我得到了它.

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector)
{
    return source.TimeInterval()
                 .Scan(Tuple.Create(TimeSpan.Zero, false, default(T)), (acc, v) =>
                 {
                     if(v.Interval >= acc.Item1)
                     {
                         return Tuple.Create(intervalSelector(v.Value), true, v.Value);
                     }
                     return Tuple.Create(acc.Item1 - v.Interval, false, v.Value);
                 })
                 .Where(t => t.Item2)
                 .Select(x => x.Item3);
}

这可以按我的意愿工作:每次产生一个值时x,它会停止产生值,直到intervalSelector(x)时间过去.