反应式扩展是否支持滚动缓冲区？

Question

我正在使用反应式扩展将数据整理到 100 毫秒的缓冲区中：

this.subscription = this.dataService
    .Where(x => !string.Equals("FOO", x.Key.Source))
    .Buffer(TimeSpan.FromMilliseconds(100))
    .ObserveOn(this.dispatcherService)
    .Where(x => x.Count != 0)
    .Subscribe(this.OnBufferReceived);

这工作得很好。但是，我想要的行为与 Buffer 操作提供的行为略有不同。本质上，如果收到另一个数据项，我想重置计时器。只有当整个 100ms 都没有收到数据时我才想处理它。这开启了永远处理数据的可能性，因此我还应该能够指定最大计数。我会想象这样的事情：

.SlidingBuffer(TimeSpan.FromMilliseconds(100), 10000)

我环顾四周，但在 Rx 中找不到类似的东西？有人能证实/否认这一点吗？

Answer 1

这可以通过组合 可观察。首先，让我们解决忽略最大计数条件的更简单的问题：

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay)
{
    var closes = stream.Throttle(delay);
    return stream.Window(() => closes).SelectMany(window => window.ToList());
}

强大的 Window 方法 完成了繁重的工作。现在很容易了解如何添加最大计数：

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? max=null)
{
    var closes = stream.Throttle(delay);
    if (max != null)
    {
        var overflows = stream.Where((x,index) => index+1>=max);
        closes = closes.Merge(overflows);
    }
    return stream.Window(() => closes).SelectMany(window => window.ToList());
}

我将在我的博客上写一篇文章对此进行解释。 https://gist.github.com/2244036

Window 方法的文档：

• http://leecampbell.blogspot.co.uk/2011/ 03/rx-part-9join-window-buffer-and-group.html
• http://enumeratethis.com/2011/07/26/financial-charts-reactive-extensions/< /a>

Answer 2

我编写了一个扩展来完成您所要做的大部分事情 - BufferWithInactivity。

这里是：

public static IObservable<IEnumerable<T>> BufferWithInactivity<T>(
    this IObservable<T> source,
    TimeSpan inactivity,
    int maximumBufferSize)
{
    return Observable.Create<IEnumerable<T>>(o =>
    {
        var gate = new object();
        var buffer = new List<T>();
        var mutable = new SerialDisposable();
        var subscription = (IDisposable)null;
        var scheduler = Scheduler.ThreadPool;

        Action dump = () =>
        {
            var bts = buffer.ToArray();
            buffer = new List<T>();
            if (o != null)
            {
                o.OnNext(bts);
            }
        };

        Action dispose = () =>
        {
            if (subscription != null)
            {
                subscription.Dispose();
            }
            mutable.Dispose();
        };

        Action<Action<IObserver<IEnumerable<T>>>> onErrorOrCompleted =
            onAction =>
            {
                lock (gate)
                {
                    dispose();
                    dump();
                    if (o != null)
                    {
                        onAction(o);
                    }
                }
            };

        Action<Exception> onError = ex =>
            onErrorOrCompleted(x => x.OnError(ex));

        Action onCompleted = () => onErrorOrCompleted(x => x.OnCompleted());

        Action<T> onNext = t =>
        {
            lock (gate)
            {
                buffer.Add(t);
                if (buffer.Count == maximumBufferSize)
                {
                    dump();
                    mutable.Disposable = Disposable.Empty;
                }
                else
                {
                    mutable.Disposable = scheduler.Schedule(inactivity, () =>
                    {
                        lock (gate)
                        {
                            dump();
                        }
                    });
                }
            }
        };

        subscription =
            source
                .ObserveOn(scheduler)
                .Subscribe(onNext, onError, onCompleted);

        return () =>
        {
            lock (gate)
            {
                o = null;
                dispose();
            }
        };
    });
}

Answer 3

使用 Rx Extensions 2.0，您可以通过接受超时和大小的新缓冲区重载来满足这两个要求：

this.subscription = this.dataService
    .Where(x => !string.Equals("FOO", x.Key.Source))
    .Buffer(TimeSpan.FromMilliseconds(100), 1)
    .ObserveOn(this.dispatcherService)
    .Where(x => x.Count != 0)
    .Subscribe(this.OnBufferReceived);

请参阅 https://msdn.microsoft.com/en-us/library/hh229200(v=vs.103).aspx 查看文档。

Answer 4

正如 Rohit Sharma 在 Panic 上校的解决方案中的评论中提到的，项目将被缓冲并且不会被缓冲的位置存在问题。除非生成了项目，否则推送给订阅者。

正如此 comment 中所述，问题是 p.Window (() => closes)，因为它打开了一个可能会错过事件的间隙。

处理每个窗口后将调用该 lambda。 Window 运算符将针对 lambda 每次返回的内容调用 Subscribe，因为据它所知，您可能每次都会从该 lambda 返回完全不同的 IObservable。

由于现在始终使用相同的 lambda，因此我们需要调整 maxCount。如果没有更改，maxCount 将永远不会重置，并且在命中一次后，每个新事件都将超过 maxCount。

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? maxCount = null)
{
    var publish = stream.Publish(p =>
    {
        var closes = p.Throttle(delay);

        if (maxCount != null)
        {
            Int32 i = 0;

            var overflows = p.Where(x =>
            {
                ++i;

                if (i >= maxCount)
                {
                    i = 0;
                    return true;
                }

                return false;
            });

            closes = closes.Merge(overflows);
        }

        return p.Window(closes).SelectMany(window => window.ToList());
    });

    return publish;
}

更新：
经过进一步的测试，我发现在某些情况下，项目仍然无法正确推送给订阅者。

以下是我们使用的解决方法，4 个月以来一直没有出现任何问题。

解决方法是将 .Delay(...) 添加到任何 TimeSpan 中。

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? maxCount = null)
{
    var publish = stream.Publish(p =>
    {
        var closes = p.Throttle(delay);

        if (maxCount != null)
        {
            var overflows = stream.Where((x, index) => index + 1 >= maxCount);
            closes = closes.Merge(overflows);
        }

        return p.Window(() => closes).SelectMany(window => window.ToList()).Delay(TimeSpan.Zero);
    });

    return publish;
}

Answer 5

我想这可以在 Buffer 方法之上实现，如下所示：

public static IObservable<IList<T>> SlidingBuffer<T>(this IObservable<T> obs, TimeSpan span, int max)
        {
            return Observable.CreateWithDisposable<IList<T>>(cl =>
            {
                var acc = new List<T>();
                return obs.Buffer(span)
                        .Subscribe(next =>
                        {
                            if (next.Count == 0) //no activity in time span
                            {
                                cl.OnNext(acc);
                                acc.Clear();
                            }
                            else
                            {
                                acc.AddRange(next);
                                if (acc.Count >= max) //max items collected
                                {
                                    cl.OnNext(acc);
                                    acc.Clear();
                                }
                            }
                        }, err => cl.OnError(err), () => { cl.OnNext(acc); cl.OnCompleted(); });
            });
        }

注意：我还没有测试过它，但我希望它能给你这个想法。

Answer 6

Panic 上校的解决方案几乎是完美的。唯一缺少的是 Publish 组件，以便使该解决方案也适用于冷序列。

/// <summary>
/// Projects each element of an observable sequence into a buffer that's sent out
/// when either a given inactivity timespan has elapsed, or it's full,
/// using the specified scheduler to run timers.
/// </summary>
public static IObservable<IList<T>> BufferUntilInactive<T>(
    this IObservable<T> source, TimeSpan dueTime, int maxCount,
    IScheduler scheduler = default)
{
    if (maxCount < 1) throw new ArgumentOutOfRangeException(nameof(maxCount));
    scheduler ??= Scheduler.Default;
    return source.Publish(published =>
    {
        var combinedBoundaries = Observable.Merge
        (
            published.Throttle(dueTime, scheduler),
            published.Skip(maxCount - 1)
        );

        return published
            .Window(() => combinedBoundaries)
            .SelectMany(window => window.ToList());
    });
}

除了添加 Publish 之外，我还用等效但更短的内容替换了原始的 .Where((_, index) => index + 1 >= maxCount) .Skip(maxCount - 1)。为了完整起见，还有一个 IScheduler 参数，用于配置运行计时器的调度程序。