Skip to content

Instantly share code, notes, and snippets.

@ronnieoverby
Created March 15, 2015 14:15
Show Gist options
  • Select an option

  • Save ronnieoverby/30e5e3f6bcb312a0bc43 to your computer and use it in GitHub Desktop.

Select an option

Save ronnieoverby/30e5e3f6bcb312a0bc43 to your computer and use it in GitHub Desktop.
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace CoreTechs.Common
{
public class BufferedEnumerable<T> : IEnumerable<T>
{
private readonly IEnumerable<T> _source;
private readonly int _bufferSize;
public BufferedEnumerable(IEnumerable<T> source, int bufferSize = 2)
{
if (bufferSize < 2)
throw new ArgumentOutOfRangeException("bufferSize", "bufferSize must not be less than 2");
_source = source;
_bufferSize = bufferSize;
}
public IEnumerator<T> GetEnumerator()
{
using (var buffer = new BlockingCollection<T>(_bufferSize))
using (var producer = Task.Run(() =>
{
try
{
foreach (var item in _source)
buffer.Add(item);
}
finally
{
buffer.CompleteAdding();
}
}))
{
foreach (var item in buffer.GetConsumingEnumerable())
yield return item;
// wait to expose exceptions
producer.Wait();
}
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment