public
Last active

BufferUntilCalm method for .NET's Reactive Extensions

  • Download Gist
buffer-until-calm.cs
C#
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
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());
}
}

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.