Extension to RX to add Throttling with a buffer

ThrottleWithBuffer.cs

/*
* This code is licensed under the terms of the MIT license
*/

public static IObservable<TSource[]> ThrottleWithBuffer<TSource>(
	this IObservable<TSource> source, TimeSpan timeout)
{
	return source.ThrottleWithBuffer(timeout, TaskPoolScheduler.Default);
}

// see http://stackoverflow.com/questions/4655437/how-to-implement-buffering-with-timeout-in-rx
// Ripped off/based on the above answer to a StackOverflow question, I preferred the name
// ThrottleWithBuffer also.
// 
// Implements throttle-like functionality, but instead of only returning the last item 
// before the throttle timeout, it keeps a buffer of the items
public static IObservable<TSource[]> ThrottleWithBuffer<TSource>(this IObservable<TSource> source, 
	TimeSpan timeout, IScheduler scheduler)
{
	return Observable.Create<TSource[]>(o =>
	{
		var buffer = new SynchronizedCollection<TSource>();

		var timeoutDisposable = new MultipleAssignmentDisposable();

		Action flushBuffer = () =>
		{
			TSource[] values;
			lock (buffer.SyncRoot)
			{
				values = buffer.ToArray();
				buffer.Clear();
			}
			if (values.Length > 0)
			{
				o.OnNext(values);
			}
		};

		var sourceSubscription = source.Subscribe(
			value =>
			{
				buffer.Add(value);

				timeoutDisposable.Disposable = scheduler.Schedule(timeout, flushBuffer);
			},
			o.OnError,
			() =>
			{
				flushBuffer();
				o.OnCompleted();
			});

		return new CompositeDisposable(sourceSubscription, timeoutDisposable);
	});
}

Hey! thanks for this! what is your license for it?

Honestly I never even thought about licences to gist, but I suppose I should. So let's simply say it's MIT. I'll maybe have to sort out how I display such licensing for Gists.

Update: Okay I've added a basic license text - thanks for pointing this out to me.

Thank you ++