BufferWhile Rx Extension

BufferWhile.cs

public static IObservable<IList<TSource>> BufferWhile<TSource>(
	this IObservable<TSource> source,
	Func<TSource, bool> condition)
{
	return source.AggregateUntil(
		() => new List<TSource>(),
		(list, value) =>
		{
			list.Add(value);
			return list;
		},
		list => !condition(list[list.Count - 1]));
}

public static IObservable<TAccumulate> AggregateUntil<TSource, TAccumulate>(
	this IObservable<TSource> source,
	Func<TAccumulate> seed,
	Func<TAccumulate, TSource, TAccumulate> accumulator,
	Func<TAccumulate, bool> predicate)
{
	return Observable.CreateWithDisposable<TAccumulate>(observer =>
		{
			var accumulate = seed();
			return source.Subscribe(value =>
				{
					accumulate = accumulator(accumulate, value);

					if (predicate(accumulate))
					{
						observer.OnNext(accumulate);
						accumulate = seed();
					}
				},
				observer.OnError,
				observer.OnCompleted);
		});
}