Rx：如何緩衝事件（環形緩衝區）並且只在發生特殊事件時才刷新它們？

Question

我在C＃中使用Reactive Extensions（Rx）並希望以下面的方式過濾事件。想象一下，我有以下的鼻祖序列：

ABCDEF X GHI XĴXX KLMNO X P

我想產生以下的輸出：

EF X HI X J XX NO X

基本上，我會緩衝（油門？）事件與最大界限（在這個例子中這個界限是2），當我得到某個事件（在這種情況下事件X），我想刷新緩衝區輸出並重新開始緩衝，直到再次看到特殊事件。

我正在嘗試一些方法，但沒有任何運氣，我想應該有一個簡單的方法來完成它，我失蹤了。編輯：一個約束，是我期望得到TONS被拋棄的事件，並且只有X的幾個實例，所以在內存中保留一個具有數千個事件的緩衝區，只讀取最後2個（或20個）事件，並不是一個真正的選擇。

Answer 1

這裏是一個刺來回答我的問題如果您發現任何問題，請告訴我。

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); 

Answer 2

爲方便起見，我們需要以下兩個擴展功能：

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 3

我會捎帶上我在這裏發佈的另一個答案： 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