Tiện ích mở rộng phản ứng: Ga/Mẫu có khoảng thời gian khác nhau

Question

Tôi có một số IObservable tạo giá trị theo khoảng thời gian ngẫu nhiên và tôi muốn điều chỉnh chuỗi này. Một điều tôi đã phát hiện ra là định nghĩa "điều chỉnh" của nhà điều hành Throttle không giống như của tôi.

Throttle chỉ tạo ra các giá trị sau khoảng thời gian đã chỉ định elapses với độ im lặng (nó tạo giá trị cuối cùng được xem). Tôi nghĩ rằng điều chỉnh sẽ có nghĩa là sản xuất các giá trị tại khoảng thời gian quy định (trừ khi có sự im lặng, tất nhiên).

Giả sử, tôi mong đợi Observable.Interval(100).Select((_,i) => i).Throttle(200) để sản xuất (modulo bất kỳ vấn đề hiệu suất/thời gian nào) các số chẵn, vì tôi điều chỉnh nó thành "nửa tốc độ". Tuy nhiên, chuỗi đó không tạo ra giá trị nào cả, vì không bao giờ có khoảng thời gian im lặng 200.

Vì vậy, tôi phát hiện ra rằng hành vi "điều chỉnh" tôi muốn. Observable.Interval(100).Select((_,i) => i).Sample(200) sản xuất (một lần nữa, modulo bất kỳ vấn đề hiệu suất/thời gian) chuỗi các số chẵn.

Tuy nhiên, tôi có một vấn đề khác: khoảng thời gian thay đổi, tùy thuộc vào giá trị "được lấy mẫu" cuối cùng. Những gì tôi muốn là viết một nhà điều hành trông như thế này:

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector); 

Tham số intervalSelector tạo ra khoảng thời gian cho các mẫu tiếp theo, và các mẫu đầu tiên ... hoặc là chụp ở giá trị đầu tiên hoặc từ một tham số bổ sung , Tôi không quan tâm.

Tôi đã cố gắng viết điều này nhưng tôi đã kết thúc với một cấu trúc phức tạp lớn không hoạt động hoàn toàn đúng. Câu hỏi của tôi là, tôi có thể xây dựng điều này bằng cách sử dụng các toán tử hiện có (hay còn gọi là một lớp lót)?

Answer 1

Nhiều giờ sau, và với một số giấc ngủ trên đó, tôi hiểu rồi.

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector) 
{ 
    return source.TimeInterval() 
       .Scan(Tuple.Create(TimeSpan.Zero, false, default(T)), (acc, v) => 
       { 
        if(v.Interval >= acc.Item1) 
        { 
         return Tuple.Create(intervalSelector(v.Value), true, v.Value); 
        } 
        return Tuple.Create(acc.Item1 - v.Interval, false, v.Value); 
       }) 
       .Where(t => t.Item2) 
       .Select(x => x.Item3); 
} 

này hoạt động như tôi muốn: mỗi lần nó tạo ra một giá trị x, nó dừng sản xuất giá trị cho đến khi intervalSelector(x) thời gian trôi qua.

Answer 2

Không phải những gì bạn đang tìm kiếm for Observable.BufferWithTime?