Архивирование с бесконечной последовательностью, которая является истинным, то всегда ложно

Question

я сделал метод расширения:

public static IObservable<T> RateLimit<T>(this IObservable<T> source, 
              TimeSpan minDelay) 
{ 
    return 
     source.TimeInterval() 
      .Select(
       (x, i) => 
        Observable.Return(x.Value) 
         .Delay(i == 0 
          ? TimeSpan.Zero 
          : TimeSpan.FromTicks(
            Math.Max(minDelay.Ticks - x.Interval.Ticks, 0)))) 
      .Concat(); 
} 

это создает новый наблюдаемыми, что позволяет только элементы через с минимальным отрывом во времени.

Чтобы удалить начальную задержку, необходимо обработать первый элемент по-разному.

Как видно, есть тест, чтобы узнать, имеем ли мы дело с первым предметом путем тестирования i == 0. Проблема здесь в том, что если мы обработаем более int.MaxValue элементов, это не удастся.

Вместо этого, я думал о следующей последовательности

var trueThenFalse = Observable.Return(true) 
        .Concat(Observable.Repeat(false)) 

и сжать его вместе с моим источником:

source.TimeInterval().Zip(trueThenFalse, ... 

, но при прохождении этой бесконечной последовательности в Zip, мы оказываемся ввести жесткий цикл где trueThenFalse испускает все предметы за один раз (до бесконечности). Потерпеть неудачу.

Я мог бы легко кодировать вокруг этого с побочными эффектами (например, bool), но это будет представлять потерю чистоты, которой я не был бы доволен.

Любые предложения?

EDIT

Хотя не совсем то же самое поведение, следующий код демонстрирует некоторые неприятные черты

var trueThenFalse = Observable.Return(true) 
    .Concat(Observable.Repeat(false)); 
var src = Observable.Interval(TimeSpan.FromSeconds(1)); //never completes 
src.Zip(trueThenFalse,(i,tf)=>tf).ForEach(x=>Trace.TraceInformation("{0}",x)); 

и в конце концов умирает с OOME. Это связано с тем, что trueThenFalse, как представляется, отключает все свои значения, но они не потребляются Zip своевременно.

Answer 1

Таким образом, получается, что Zip имеет another overload, который может объединять последовательность IObservable с последовательностью IEnumerable.

Объединив семантику ввода IObservable с семантикой вытягивания IEnumerable, можно получить мой тестовый пример.

Таким образом, с помощью следующего метода:

private IEnumerable<T> Inf<T>(T item) 
{ 
    for (;;) 
    { 
     yield return item; 
    } 
} 

мы можем сделать IEnumerable:

var trueThenFalse = Enumerable.Repeat(true, 1).Concat(Inf(false)); 

, а затем Zip его с источником наблюдаемого:

var src = Observable.Interval(TimeSpan.FromSeconds(1)); 
src.Zip(trueThenFalse, (i, tf) => tf).ForEach(x => Trace.TraceInformation("{0}", x)); 

... и все работает так, как ожидалось.

я теперь следующая реализация моего метода RateLimiter:

public static IObservable<T> RateLimit<T>(this IObservable<T> source, 
              TimeSpan minDelay) 
{ 
    var trueThenFalse = Enumerable.Repeat(true, 1).Concat(Inf(false)); 
    return 
     source.TimeInterval() 
      .Zip(trueThenFalse, (item, firstTime) => Observable.Return(item.Value) 
       .Delay(firstTime 
        ? TimeSpan.Zero 
        : TimeSpan.FromTicks(
         Math.Max(minDelay.Ticks - item.Interval.Ticks, 0)))) 

      .Concat(); 
} 

Answer 2

Это похоже на Rx IObservable buffering to smooth out bursts of events, хотя вы явно пытаетесь понять, почему ваше решение делает/не работает.

Я нахожу решение там более элегантным, хотя каждому свой.