How can I alternately buffer and flow a live data stream in Rx

by

I would do this as follows:

  • Window the data stream using the gate stream as a closing selector
  • We can use DistinctUntilChanged on the gate stream to ensure no repeated values
  • We will also force the gate stream to start closed (false) – it won’t affect the output and allows a neat trick
  • Then use the overload of Select that gives each element an index number. With this we can tell if we need to buffer or just emit the window as is, because we know that even-numbered windows are for buffering (because we made sure the gate stream starts with false)
  • We can use ToList() to buffer each even window until it closes – this is the actually equivalent of a Buffer() that waits until OnCompleted
  • We use an identity SelectMany to flatten the buffered windows
  • Finally we concatenate the windows in order to guarantee order is preserved

It looks like this:

dataStream.Window(gateStream.StartWith(false).DistinctUntilChanged())
          .Select((w, i) =>  i % 2 == 0 ? w.ToList().SelectMany(x => x) : w)
          .Concat()
          .Subscribe(Console.WriteLine);

More Related Contents:

  1. ObserveOn and SubscribeOn – where the work is being done
  2. How to throttle event stream using RX?
  3. How can I implement an exhaustMap handler in Rx.Net?
  4. Why are Subjects not recommended in .NET Reactive Extensions?
  5. How can I see what my reactive extensions query is doing?
  6. Write an Rx “RetryAfter” extension method
  7. Good introduction to the .NET Reactive Framework [closed]
  8. With Rx, how do I ignore all-except-the-latest value when my Subscribe method is running
  9. Does reactive extensions support rolling buffers?
  10. Rx: How can I respond immediately, and throttle subsequent requests
  11. A way to push buffered events in even intervals
  12. What are the default Schedulers for each observable operator?
  13. What’s a good way to run periodic tasks using Rx, with a single concurrent execution restriction?
  14. How to call back async function from Rx Subscribe?
  15. Reactive Extensions for .NET (Rx): Take action once all events are completed
  16. Pause and Resume Subscription on cold IObservable
  17. How to check if element exist or not without using try-catch
  18. How can I update the current line in a C# Windows Console App?
  19. How to serialize a Dictionary as part of its parent object using Json.Net
  20. Lifetime of ASP.NET Static Variable
  21. How to enable design support in a custom control?
  22. FTPS (FTP over SSL) in C#
  23. Drawing glitches when using CreateGraphics rather than Paint event handler for custom drawing
  24. Fast creation of objects instead of Activator.CreateInstance(type)
  25. Performance Cost Of ‘try’ in C#
  26. Print the source filename and linenumber in C#
  27. PowerShell – How to Import-Module in a Runspace
  28. Using a Scoped service in a Singleton in an Asp.Net Core app
  29. Error while trying to run project:The module was expected to contain an assembly manifest
  30. Why does Console.In.ReadLineAsync block?

Leave a Comment