Rx.NET의 RxJS throttleTime과 같습니다.

Question

Rx.NET을 사용하여 RxJS throttleTime을 구현하려고합니다 (.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 변수)을 유지합니다. 터플의 다른 필드는 더 많은 다운 스트림에 도움이됩니다.