Buffer & Window

Buffer

Buffer has got number of overloads to suit your needs. All it does is chops the stream into windows and returns observable of lists of elements. Lets start with the following example and dig down later:-


        static void Main(string[] args)
        {
           var input = new[] {1,2,3,4,5}.ToObservable();
            int i = 0;
            input.Buffer(2).Subscribe(list => Console.WriteLine(list.Sum()));
            Console.ReadKey();

        }

Window
Window returns <IObservable<IObservable<T>>> which make it a bit complex, but this method is extremely useful. Unlike the Buffer, which is caching the item internally, the Window does not cache the items at all, each item is immediately project through IObservable<T>.OnNext.

Following Marble diagram will show the differences between two operators:-

Following is an example of using a window to find Max prices. The window is non-overlapping one containing five elements each and pumping out maximum for each window. If you are familiar to candlestick charts, this is the same thing as the high of each candlestick.


    class Program
    {
        static void Main(string[] args)
        {
            var input = GetPrice().ToObservable();

            var windowMax = from window in input.Window(2, 1)
                                   from item in window.Max()
                                   select item;

            windowMax.Subscribe(Console.WriteLine);
   
            Console.ReadKey();
        }

        private static IEnumerable<double> GetPrice()
        {
            var r = new Random(DateTime.Now.Millisecond);

            while (true)
            {
                Thread.Sleep(500);
                yield return Math.Round(r.NextDouble(), 2);
            }
        }

    }



The output:-