CSharp · Console
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Runtime.CompilerServices;
public static class Program
{
// Was having issues with the NuGet package so wrote a custom Range function instead, at the bottom
public static async Task Main()
{
var sequence_A = Range("A{0}", 1, 5); // note the DIFFERENT size ranges
var sequence_B = Range("B{0}", 1, 3);
var sequence_C = Range("C{0}", 1, 7);
var merged = Interleave(sequence_A, sequence_B, sequence_C);
await foreach (var item in merged) { Console.WriteLine(item); await Task.Delay(100); }
}
public static IAsyncEnumerable<T> Interleave<T>(params IAsyncEnumerable<T>[] sources) =>
Interleave(default, sources);
public static async IAsyncEnumerable<T> Interleave<T>([EnumeratorCancellation] CancellationToken token, IAsyncEnumerable<T>[] sources)
{
if(sources.Length == 0)
yield break;
var enumerators = new List<(IAsyncEnumerator<T> e, Task<bool> t)>(sources.Length);
try
{
for(var i = 0; i < sources.Length; i++)
{
var e = sources[i].GetAsyncEnumerator(token);
enumerators.Add((e, e.MoveNextAsync().AsTask()));
}
do
{
var taskResult = await Task.WhenAny(enumerators.Select(tuple => tuple.t));
var ind = enumerators.FindIndex(tuple => tuple.t == taskResult);
if(ind == -1) throw new Exception("ind was "+ind);
var tuple = enumerators[ind];
enumerators.RemoveAt(ind);
if(taskResult.Result)
{
yield return tuple.e.Current;
enumerators.Add((tuple.e, tuple.e.MoveNextAsync().AsTask()));
}
else
{
try
{
await tuple.e.DisposeAsync();
}
catch
{ //
}
}
} while (enumerators.Count > 0);
}
finally
{
for(var i = 0; i < enumerators.Count; i++)
{
try
{
await enumerators[i].e.DisposeAsync();
}
catch
{ //
}
}
}
}
static async IAsyncEnumerable<string> Range(string text, int start, int count) {
for (int i = 0; i < count; i++)
{
var delayTask = Task.Delay(50);
yield return string.Format(text, start + i);
await delayTask;
}
}
}