Tiện ích mở rộng phản ứng có hỗ trợ bộ đệm lăn không?

Question

Tôi đang sử dụng phần mở rộng phản ứng để đối chiếu dữ liệu vào bộ đệm của 100ms:

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

này hoạt động tốt. Tuy nhiên, tôi muốn có hành vi hơi khác so với hành vi của hoạt động Buffer. Về cơ bản, tôi muốn đặt lại bộ hẹn giờ nếu nhận được một mục dữ liệu khác. Chỉ khi không có dữ liệu nào được nhận cho toàn bộ 100ms thì tôi có muốn xử lý nó không. Điều này mở ra khả năng không bao giờ xử lý dữ liệu, vì vậy tôi cũng có thể chỉ định số lượng tối đa. Tôi sẽ tưởng tượng điều gì đó dọc theo các dòng:

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

Tôi đã xem xét và không thể tìm thấy bất kỳ thứ gì như thế này trong Rx? Bất cứ ai có thể xác nhận/từ chối điều này?

Answer 1

Tôi đã viết phần mở rộng để thực hiện hầu hết những gì bạn đang theo dõi - BufferWithInactivity.

Ở đây là:

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 2

Tôi đoán đây có thể được thực hiện trên phương pháp đệm như hình dưới đây:

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

Chú ý: Tôi đã không kiểm tra nó, nhưng tôi hy vọng nó cho bạn ý tưởng.

Answer 3

Điều này có thể bằng cách kết hợp các phương pháp được xây dựng trong Window và Throttle của Observable. Trước tiên, chúng ta hãy giải quyết vấn đề đơn giản hơn khi chúng ta bỏ qua điều kiện đếm tối đa:

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

Mạnh mẽ nâng lên. Giờ đây, thật dễ dàng để xem cách thêm số lượng tối đa:

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

Tôi sẽ viết một bài giải thích điều này trên blog của mình. https://gist.github.com/2244036

Tài liệu dành cho phương pháp cửa sổ:

• 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/

Answer 4

Với Rx Extensions 2.0, bạn có thể trả lời tất cả các yêu cầu với một bộ đệm mới quá tải chấp nhận một thời gian chờ và kích thước:

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

Xem https://msdn.microsoft.com/en-us/library/hh229200(v=vs.103).aspx để biết tài liệu.