Rx:如何缓冲事件(环缓冲区)，并仅在特殊事件发生时刷新它们

Question

我在c#中使用响应式扩展(Rx)，并希望以以下方式过滤事件。假设我有以下的始发者序列:

B E C D F X G H X J X K L M N O X p>

想产生以下输出:

E F X H我 X J 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

我将借用我在这里发布的另一个答案:在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);