Does reactive extensions support rolling buffers?

by

This is possible by combining the built-in Window and Throttle methods of Observable. First, let’s solve the simpler problem where we ignore the maximum count condition:

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay)
{
    var closes = stream.Throttle(delay);
    return stream.Window(() => closes).SelectMany(window => window.ToList());
}

The powerful Window method did the heavy lifting. Now it’s easy enough to see how to add a maximum count:

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? max=null)
{
    var closes = stream.Throttle(delay);
    if (max != null)
    {
        var overflows = stream.Where((x,index) => index+1>=max);
        closes = closes.Merge(overflows);
    }
    return stream.Window(() => closes).SelectMany(window => window.ToList());
}

I’ll write a post explaining this on my blog. https://gist.github.com/2244036

Documentation for the Window method:

More Related Contents:

  1. ObserveOn and SubscribeOn – where the work is being done
  2. How can I implement an exhaustMap handler in Rx.Net?
  3. Good introduction to the .NET Reactive Framework [closed]
  4. A way to push buffered events in even intervals
  5. Reactive Extensions for .NET (Rx): Take action once all events are completed
  6. Easiest way to compare arrays in C#
  7. Difference between ObservableCollection and BindingList
  8. XML serialization of interface property
  9. How do arrays in C# partially implement IList?
  10. String Concatenation using ‘+’ operator
  11. How to save a List on Settings.Default?
  12. Escape invalid XML characters in C#
  13. Can I get command line arguments of other processes from .NET/C#?
  14. What is the fastest way to create a checksum for large files in C#
  15. How to set system environment variable in C#?
  16. DateTime to javascript date
  17. Should we always include a default constructor in the class?
  18. How to extend IdentityUser with custom property
  19. How to concatenate two IEnumerable into a new IEnumerable?
  20. Travel through pixels in BMP
  21. memcpy function in c# [duplicate]
  22. .NET Core vs Mono
  23. Validating file types by regular expression
  24. How to do an “in” query in entity framework?
  25. ASP.NET Web API and [Serializable] class
  26. trying to convert rgb from a .net Color to a string such as “red” or “blue”
  27. How to include() nested child entity in linq
  28. Get the calling function name from the called function [duplicate]
  29. What ‘length’ parameter should I pass to SqlDataReader.GetBytes()
  30. C# generic methods, type parameters in new() constructor constraint

Leave a Comment