Create a gist now

Instantly share code, notes, and snippets.

@v2m /gist:4969559
Last active Dec 13, 2015

Execute implementation
public IEnumerable<Row> Execute(IEnumerable<Row> rows)
{
var blockingCollection = new BlockingCollection<Row>();
var count = _operations.Count;
if (count == 0)
{
yield break;
}
var tasks = _operations.Select(currentOp =>
Task.Factory.StartNew(() =>
{
try
{
foreach (var row in currentOp.Execute(null))
{
blockingCollection.Add(row);
}
}
finally
{
if (Interlocked.Decrement(ref count) == 0)
{
blockingCollection.CompleteAdding();
}
}
}
)
).ToArray();
foreach (var row in blockingCollection.GetConsumingEnumerable())
{
yield return row;
}
Task.WaitAll(tasks); // raise any exception that were raised during execution
}
@dalenewman

Works great. Thanks

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment