Có thể để Observable.Buffer trên một cái gì đó khác với thời gian

Question

Tôi đã tìm kiếm các ví dụ về cách sử dụng Observable.Buffer trong rx nhưng không thể tìm thấy bất cứ điều gì đáng kể hơn so với các công cụ đệm thời gian trên đĩa.

Dường như có quá tải để chỉ định "bufferClosingSelector" nhưng tôi không thể che giấu suy nghĩ của mình xung quanh nó.

Những gì tôi đang cố gắng làm là tạo chuỗi có bộ đệm theo thời gian hoặc bằng cách "tích lũy". Hãy xem xét luồng yêu cầu trong đó mọi yêu cầu có trọng lượng nào đó và tôi không muốn xử lý nhiều hơn x trọng lượng tích lũy tại một thời điểm hoặc nếu chưa đủ tích lũy, hãy cho tôi biết những gì đã đến trong khung thời gian trước (Bộ đệm thông thường) chức năng)

Answer 1

bufferClosingSelector là một hàm được gọi là mọi thời gian để có được một Observable sẽ tạo ra một giá trị khi bộ đệm được dự kiến ​​đóng lại.

Ví dụ,

source.Buffer(() => Observable.Timer(TimeSpan.FromSeconds(1))) công trình như Buffer(time) quá tải thường xuyên.

Trong trường hợp bạn muốn cân một chuỗi, bạn có thể áp dụng Scan theo trình tự và sau đó quyết định điều kiện tổng hợp của mình.

Ví dụ, source.Scan((a,c) => a + c).SkipWhile(a => a < 100) mang đến cho bạn một chuỗi trong đó sản xuất một giá trị khi chuỗi nguồn đã bổ sung lên đến hơn 100.

Bạn có thể sử dụng để đua Amb hai điều kiện đóng cửa để xem những phản ứng đầu tiên:

 .Buffer(() => Observable.Amb 
        (
          Observable.Timer(TimeSpan.FromSeconds(1)), 
          source.Scan((a,c) => a + c).SkipWhile(a => a < 100) 
        ) 
       ) 

Bạn có thể sử dụng bất kỳ bộ kết hợp nào tạo ra bất kỳ giá trị nào cho bộ đệm được đóng tại điểm đó.

Lưu ý: Giá trị được cung cấp cho bộ chọn đóng không quan trọng - đó là thông báo quan trọng. Vì vậy, để kết hợp các nguồn khác nhau với Amb, chỉ cần thay đổi nó thành System.Reactive.Unit.

Observable.Amb(stream1.Select(_ => new Unit()), stream2.Select(_ => new Unit())