Created

Embed URL

HTTPS clone URL

SSH clone URL

You can clone with HTTPS or SSH.

Download Gist

BufferUntilCalm method for .NET's Reactive Extensions

View buffer-until-calm.cs
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());
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.