Rx：如何缓冲事件（环形缓冲区）并仅在发生特殊事件时刷新它们？

Question

我在 C# 中使用 Reactive Extensions (Rx)，并希望通过以下方式过滤事件。想象一下，我有以下发起者序列：

A B C D E F X G H I X J X X K L M N O X P

我想产生以下输出：

E F X H I X J X X N O X

基本上，我会缓冲（节流？）具有最大界限（在示例中此界限为 2）的事件，当我得到某个事件（在本例中为事件 X）时，我想将该缓冲区刷新到输出并再次开始缓冲，直到我再次看到特殊事件。

我正在尝试几种方法，但没有任何运气，我想应该有一种简单的方法来完成它，我错过了。

编辑：一个限制是，我希望获得大量被丢弃的事件，并且只有少数 X 实例，因此在内存中保留一个包含数千个事件的缓冲区以仅读取最后 2 个（或 20 个）是不是一个真正的选择。

Answer 1

为方便起见，我们需要以下两个扩展函数：

public static class Extensions
{
    public static IObservable<IList<TSource>> BufferUntil<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
    {
        var published = source.Publish().RefCount();
        return published.Buffer(() => published.Where(predicate));
    }

    public static IEnumerable<TSource> TakeLast<TSource>(this IEnumerable<TSource> source, int count)
    {
        return source.Reverse().Take(count).Reverse();
    }
}

然后我们这样解决问题：

source.BufferUntil(c => c == 'X')
    .SelectMany(list => list.TakeLast(3))

输出：

E F X H I X J X X N O X

Answer 2

我会捎带我在这里发布的另一个答案： Trouble Implementing a Sliding Window in Rx

重要的是这个扩展方法：

public static class Ext
{
    public static IObservable<IList<T>> SlidingWindow<T>(
        this IObservable<T> src, 
        int windowSize)
    {
        var feed = src.Publish().RefCount();    
        // (skip 0) + (skip 1) + (skip 2) + ... + (skip nth) => return as list  
        return Observable.Zip(
        Enumerable.Range(0, windowSize)
            .Select(skip => feed.Skip(skip))
            .ToArray());
    }
}

你可以这样使用：

void Main()
{
    // A faked up source
    var source = new Subject<char>();

    var bufferSize = 2;
    Func<char, bool> eventTrigger = c => c == 'X';

    var query = source
        .Publish()
        .RefCount()
        // Want one extra slot to detect the "event"
        .SlidingWindow(bufferSize + 1)
        .Where(window => eventTrigger(window.Last()))
        .Select(buffer => buffer.ToObservable())
        .Switch();

    using(query.Subscribe(Console.WriteLine))
    {
        source.OnNext('A');
        source.OnNext('B');
        source.OnNext('C');
        source.OnNext('D');
        source.OnNext('E');
        source.OnNext('F');
        source.OnNext('X');
        source.OnNext('G');
        source.OnNext('H');
        source.OnNext('I');
        source.OnNext('X');
        Console.ReadLine();
    }    
}

输出：

E
F
X
H
I
X

Answer 3

这是一个回答我自己问题的刺，如果您发现任何问题，请告诉我。

public static class ObservableHelper
{
    /// <summary>
    /// Buffers entries that do no satisfy the <paramref name="shouldFlush"/> condition, using a circular buffer with a max
    /// capacity. When an entry that satisfies the condition ocurrs, then it flushes the circular buffer and the new entry,
    /// and starts buffering again.
    /// </summary>
    /// <typeparam name="T">The type of entry.</typeparam>
    /// <param name="stream">The original stream of events.</param>
    /// <param name="shouldFlush">The condition that defines whether the item and the buffered entries are flushed.</param>
    /// <param name="bufferSize">The buffer size for accumulated entries.</param>
    /// <returns>An observable that has this filtering capability.</returns>
    public static IObservable<T> FlushOnTrigger<T>(this IObservable<T> stream, Func<T, bool> shouldFlush, int bufferSize)
    {
        if (stream == null) throw new ArgumentNullException("stream");
        if (shouldFlush == null) throw new ArgumentNullException("shouldFlush");
        if (bufferSize < 1) throw new ArgumentOutOfRangeException("bufferSize");

        return System.Reactive.Linq.Observable.Create<T>(observer =>
        {
            var buffer = new CircularBuffer<T>(bufferSize);
            var subscription = stream.Subscribe(
                newItem =>
                    {
                        bool result;
                        try
                        {
                            result = shouldFlush(newItem);
                        }
                        catch (Exception ex)
                        {
                            return;
                        }

                        if (result)
                        {
                            foreach (var buffered in buffer.TakeAll())
                            {
                                observer.OnNext(buffered);
                            }

                            observer.OnNext(newItem);
                        }
                        else
                        {
                            buffer.Add(newItem);
                        }
                    },
                observer.OnError,
                observer.OnCompleted);

            return subscription;
        });
    }
}

顺便说一句，CircularBuffer 不是开箱即用的，但实现很简单。

那我就打个电话：

        data
            .FlushOnTrigger(item => item == 'X', bufferSize: 2)
            .Subscribe(Console.WriteLine);