将窗口拆分为窗口

Question

我在寻找Observable.Window like opener with opening window选择器

例如，我们可以采用一系列自然数。 我想知道如何将该序列分割成窗口，以便每一个新的窗口开始，如果数目大于4或窗口大小已经达到3

输入序列IObservable<int> 输出序列是IObservable<IObservable<int>>

序列1 2 5 3 1 1 2 3 1 5 0 将产生窗户1 2; 5 3 1; 1 2 3; 1; 5 0

Answer 1

使用C＃这样工作的：

var observable = 
    source 
     .Concat(Observable.Return(-1)) 
     .Publish(sp => 
      sp.Zip(sp.Skip(1), (x0, x1) => new { x0, x1 }) 
       .Publish(zsp => 
        zsp 
         .Window(zsp.Where(x => x.x1 >= 4)) 
         .Select(xs => xs.Select(x => x.x0).Window(3)) 
         .Merge())); 

我得到这样的结果：

Answer 2

按我对我不确定序列1, 2, 3, 5是否应该产生的窗口的问题评论1 2 3 | 5或1 2 3 | | 5（请参阅Enigmativity答案产生的空窗口）。我的回答不会产生空窗：

public static IObservable<IObservable<T>> Window<T>(this IObservable<T> source, Func<T, bool> predicate, int maximumWindowSize) 
{ 
    return Observable.Create<IObservable<T>>(obs => 
    { 
     var currentWindow = new Subject<T>(); 
     obs.OnNext(currentWindow); 

     var count = 0; 

     return source.Subscribe(x => 
     { 
      if (count == maximumWindowSize || predicate(x)) 
      { 
       count = 0; 
       currentWindow.OnCompleted(); 
       currentWindow = new Subject<T>(); 
       obs.OnNext(currentWindow); 
      } 
      currentWindow.OnNext(x); 
      count++; 
     }, obs.OnError,() => 
     { 
      obs.OnCompleted(); 
      currentWindow.OnCompleted(); 
     }); 
    }); 
} 

这可以用于像这样：

var windows = source.Window(x => x > 4, 3);