BufferUntilCalm method for .NET's Reactive Extensions

buffer-until-calm.cs

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());
	}
}