Расщепление Наблюдаемые в окна

Question

я ищу Observable.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);