I am trying to write A C# version of a method in the Java Stream API called Collectors.teeing() (It's like Aggregate but with multiple accumulator functions), just for educational purposes.
I want to make sure that I:
- don't enumerate the source more than once (because that is trivial)
- don't store the elements from the source (that's also trivial) meaning each element has to be processed before moving on to the next.
- don't have to reimplement anything. In other words, if for example I want to find the min and max values, I should just be able to call Min() and Max() and not have to write a loop.
This is what I came up with:
public static class Extensions
{
public static (T1, T2) Tee<TSource, T1, T2>(
this IEnumerable<TSource> source,
Func<IEnumerable<TSource>, T1> accumulator1,
Func<IEnumerable<TSource>, T2> accumulator2)
{
var tee = new Teeing<TSource>(source, 2);
Task<T1> t1 = Task.Run(() => accumulator1(tee));
Task<T2> t2 = Task.Run(() => accumulator2(tee));
Task.WaitAll(t1, t2);
return (t1.Result, t2.Result);
}
//overloads with more parameters. T3, T4, etc
private class Teeing<T> : IEnumerable<T>, IEnumerator<T>
{
private IEnumerator<T> sourceEnumerator;
private Barrier barrier;
private bool hasNext;
public Teeing(IEnumerable<T> source, int count)
{
sourceEnumerator = source.GetEnumerator();
barrier = new Barrier(count, b => {
hasNext = sourceEnumerator.MoveNext();
});
}
public T Current => sourceEnumerator.Current;
public bool MoveNext()
{
barrier.SignalAndWait();
return hasNext;
}
public void Dispose()
{
barrier.RemoveParticipant();
if (barrier.ParticipantCount == 0) {
barrier.Dispose();
sourceEnumerator.Dispose();
}
}
public IEnumerator<T> GetEnumerator() => this;
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
object IEnumerator.Current => Current!;
void IEnumerator.Reset() => throw new NotImplementedException();
}
}
then call:
var tuple = sequence.Tee(
s => s.Min(),
s => s.Max()
);
Any exceptions thrown by one of the accumulator functions will be collected in an AggregateException. I guess that works. Or is there a better way?