Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
BufferUntilCalm method for .NET's Reactive Extensions
public static class ObservableExtensions
/// <summary>
/// Group observable sequence into buffers separated by periods of calm
/// </summary>
/// <param name="source">Observable to buffer</param>
/// <param name="calmDuration">Duration of calm after which to close buffer</param>
/// <param name="maxCount">Max size to buffer before returning</param>
/// <param name="maxDuration">Max duration to buffer before returning</param>
public static IObservable<IList<T>> BufferUntilCalm<T>(this IObservable<T> source, TimeSpan calmDuration, Int32? maxCount=null, TimeSpan? maxDuration = null)
if (source == null)
throw new ArgumentNullException("source");
var closes = source.Throttle(calmDuration);
if (maxCount != null)
var overflows = source.Where((x,index) => index+1 >= maxCount);
closes = closes.Amb(overflows);
if (maxDuration != null)
var ages = source.Delay(maxDuration.Value);
closes = closes.Amb(ages);
return source.Window(() => closes).SelectMany(window => window.ToList());
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment