具有無限序列是真的荏苒則始終爲假

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, ... 

但通過這個無限序列拉鍊時，我們似乎進入一個死循環其中trueThenFalse一次發出所有物品（無限）。失敗。

我可以很容易地編寫這個帶有副作用的代碼（例如在外部範圍內的bool），但這會代表我不滿意的純度損失。

有什麼建議嗎？

編輯

雖然不太一樣的行爲，下面的代碼顯示出一些討厭的特質

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)); 

，然後用觀察到的源住口：

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，但你顯然正在試圖理解爲什麼你的解決方案並/不起作用。

我覺得那裏的解決方案更優雅，儘管每個人都有自己的解決方案。