用一个无限的序列压缩，这个序列是真的然后总是假的

Question

我做了一个扩展方法：

public static IObservable<T> RateLimit<T>(this IObservable<T> source, 
                                          TimeSpan minDelay)
{
    return
        source.TimeInterval()
            .Select(
                (x, i) =>
                    Observable.Return(x.Value)
                        .Delay(i == 0
                            ? TimeSpan.Zero
                            : TimeSpan.FromTicks(
                                  Math.Max(minDelay.Ticks - x.Interval.Ticks, 0))))
            .Concat();
}

这会创建一个新的可观察对象，它只允许项目以最小的时间间隔通过。

要消除初始延迟，有必要区别对待第一项。

可以看出，有一个测试是通过测试i == 0来判断我们是否处理的是第一项。这里的问题是，如果我们处理超过 int.MaxValue 个项目，这将失败。

相反，我考虑了以下顺序

var trueThenFalse = Observable.Return(true)
                    .Concat(Observable.Repeat(false))

并将其与我的来源一起压缩：

source.TimeInterval().Zip(trueThenFalse, ...

但是当将这个无限序列传递给 Zip 时，我们似乎进入了一个紧密循环，其中 trueThenFalse 一次性发出所有项目（到无穷大）。失败。

我可以很容易地用副作用来解决这个问题（例如，在外部范围中有一个 bool），但这将代表我不满意的纯度损失。

有什么建议吗？

编辑

尽管行为不完全相同，但以下代码表现出一些令人讨厌的特征

var trueThenFalse = Observable.Return(true)
    .Concat(Observable.Repeat(false));
var src = Observable.Interval(TimeSpan.FromSeconds(1)); //never completes
src.Zip(trueThenFalse,(i,tf)=>tf).ForEach(x=>Trace.TraceInformation("{0}",x));

并最终死于 OOME。这是因为 trueThenFalse 似乎正在取消假脱机其所有值，但 Zip 没有及时消耗它们。

Answer 1

原来 Zip 有 another overload 可以将一个 IObservable 序列和一个 IEnumerable 序列压缩在一起。

通过将 IObservable 的推送语义与 IEnumerable 的拉语义相结合，可以让我的测试用例正常工作。

所以，用下面的方法：

private IEnumerable<T> Inf<T>(T item)
{
    for (;;)
    {
        yield return item;
    }
}

我们可以创建一个 IEnumerable:

var trueThenFalse = Enumerable.Repeat(true, 1).Concat(Inf(false));

然后使用可观察源压缩它：

var src = Observable.Interval(TimeSpan.FromSeconds(1));
src.Zip(trueThenFalse, (i, tf) => tf).ForEach(x => Trace.TraceInformation("{0}", x));

...一切正常。

我的 RateLimiter 方法现在有以下实现：

public static IObservable<T> RateLimit<T>(this IObservable<T> source,
                                          TimeSpan minDelay)
{
    var trueThenFalse = Enumerable.Repeat(true, 1).Concat(Inf(false));
    return
        source.TimeInterval()
            .Zip(trueThenFalse, (item, firstTime) => Observable.Return(item.Value)
                .Delay(firstTime
                    ? TimeSpan.Zero
                    : TimeSpan.FromTicks(
                        Math.Max(minDelay.Ticks - item.Interval.Ticks, 0))))

            .Concat();
}

Answer 2

这与 Rx IObservable buffering to smooth out bursts of events 类似，尽管您显然是在试图理解为什么您的解决方案 does/doesn 不起作用。

我发现那里的解决方案更优雅，尽管每个人都有自己的解决方案。

用一个无限的序列压缩，这个序列是真的然后总是假的

Zipping with an infinite sequence that is true then always false

c#

infinite

system.reactive