Rx.NETのRxJSスロットルタイムに相当する

Question

RxJS throttleTime相当のRx.NETを実装しようとしています（.NETでは実装されていません）。

throttleTimeは、Observableソースから値を出し、その後のソース値を無視して、このプロセスを繰り返します。 これは、発光するまでにある時間待つthrottleとは異なります。 throttleTimeが最初の項目を出してから、再度発光するまで待ちます。

これについてはどのように進むべきかについての指針は非常に高く評価されます。

Answer 1

これはそれを行う必要があります。

public static IObservable<T> ThrottleTime<T>(this IObservable<T> source, TimeSpan ts) 
{ 
    return ThrottleTime(source, ts, Scheduler.Default); 
} 

public static IObservable<T> ThrottleTime<T>(this IObservable<T> source, TimeSpan ts, IScheduler scheduler) 
{ 
    return source 
     .Timestamp(scheduler) 
     .Scan((EmitValue: false, OpenTime: DateTimeOffset.MinValue, Item: default(T)), (state, item) => item.Timestamp > state.OpenTime 
      ? (true, item.Timestamp + ts, item.Value) 
      : (false, state.OpenTime, item.Value) 
     ) 
     .Where(t => t.EmitValue) 
     .Select(t => t.Item); 
} 

説明：単一の状態変数を持つようThrottleTimeを考える：ゲートが新しい値に開く次回。ソースアイテムがこのタイムゲート値の前にある場合、何も起こりません。ソース項目がそれより後にある場合は、通過させ、ゲート値を最新のタイムスタンプにリセットします。

Scanは、タプルの時間値（OpenTime変数）を保持します。タプルの他のフィールドは、より下流で役立ちます。