19

我有一个 Observable 序列，它以快速爆发的方式产生事件（即：五个事件一个接一个，然后是长时间的延迟，然后是另一个快速的事件爆发，等等）。我想通过在事件之间插入短暂的延迟来消除这些突发。想象下图作为示例：

原始：--oooo--------------ooooo-----oo----ooo|
缓冲：--o--o--o--o--------o--o--o--o--o--o--o---------- o--o--o|

Observable.Interval()我目前的方法是在可以从原始流中提取另一个事件时通过该信号生成一个类似节拍器的计时器。问题是我无法弄清楚如何将该计时器与我的原始无缓冲可观察序列结合起来。

IObservable.Zip()接近于做我想做的事，但它只有在原始流比计时器更快地产生事件时才有效。一旦原始流中出现明显的停顿，计时器就会建立一系列不需要的事件，然后立即与原始流中的下一个突发事件配对。

理想情况下，我想要一个具有以下函数签名的 IObservable 扩展方法，它产生我上面概述的行为。现在，来拯救我的 StackOverflow :)

public static IObservable<T> Buffered(this IObservable<T> src, TimeSpan minDelay)

PS。我是 Rx 的新手，所以如果这是一个非常简单的问题，我深表歉意......

1. 简单但有缺陷的方法

这是我最初的天真和简单的解决方案，有很多问题：

public static IObservable<T> Buffered<T>(this IObservable<T> source, TimeSpan minDelay)
{
    Queue<T> q = new Queue<T>();
    source.Subscribe(x => q.Enqueue(x));
    return Observable.Interval(minDelay).Where(_ => q.Count > 0).Select(_ => q.Dequeue());
}

第一个明显的问题是内部订阅返回的 IDisposable 丢失了原始源，因此无法终止订阅。在此方法返回的 IDisposable 上调用 Dispose 会终止计时器，但不会终止底层原始事件源，因为它现在不必要地填充队列，没有人可以从队列中提取事件。

第二个问题是异常或流结束通知无法从原始事件流传播到缓冲流 - 在订阅原始源时它们会被简单地忽略。

最后但并非最不重要的一点是，现在我的代码会定期唤醒，无论是否实际上有任何工作要做，在这个美妙的新反应式世界中，我宁愿避免这种情况。

2. 方法过于复杂

为了解决我最初的简单方法中遇到的问题，我编写了一个更复杂的函数，其行为很像IObservable.Delay()（我使用 .NET Reflector 来读取该代码并将其用作我的函数的基础）。不幸的是，很多样板逻辑例如AnonymousObservable在 system.reactive 代码之外无法公开访问，所以我不得不复制和粘贴很多代码。该解决方案似乎可行，但鉴于其复杂性，我不太相信它没有错误。

我简直不敢相信没有办法使用标准响应式扩展的某种组合来实现这一点。我讨厌我在不必要地重新发明轮子的感觉，而我正在尝试构建的模式似乎是一个相当标准的模式。

4

1 回答 1

10

这实际上是A way to push buffered events in even interval的副本，但我将在此处包含一个摘要（原始看起来很混乱，因为它查看了一些替代方案）。

public static IObservable<T> Buffered<T>(this IObservable<T> source, TimeSpan minDelay)
{
    return source.Drain(x => 
        Observable.Empty<int>()
            .Delay(minDelay)
            .StartWith(x)
    );
}

我的 Drain 实现类似于SelectMany，除了它等待前一个输出首先完成（您可以将其视为ConactMany，而SelectMany更像MergeMany）。内置Drain不能以这种方式工作，因此您需要包含以下实现：

public static class ObservableDrainExtensions
{
    public static IObservable<TOut> Drain<TSource, TOut>(
        this IObservable<TSource> source, 
        Func<TSource, IObservable<TOut>> selector)
    {
        return Observable.Defer(() =>
        {
            BehaviorSubject<Unit> queue = new BehaviorSubject<Unit>(new Unit());

            return source
                .Zip(queue, (v, q) => v)
                .SelectMany(v => selector(v)
                    .Do(_ => { }, () => queue.OnNext(new Unit()))
                );
        });
    }
}
于 2010-12-22T09:46:47.390 回答