c# - Factory for IAsyncEnumerable or IAsyncEnumerator

Question

I'm wondering if there is a way to create either IAsyncEnumerable<T> or IAsyncEnumerator<T> via a Source object, rather like TaskCompletionSource allows one to do for tasks. In particular, TaskCompletionSource can be passed around like any other parameter.

Maybe something like this:

public class AsyncEnumerables {

    public Task HandlerTask { get; set; }

    public async Task<string> ParentMethod() {
        var source = new AsyncEnumerableSource<int>();
        IAsyncEnumerable asyncEnumerable = source.GetAsyncEnumerable();
        HandlerTask = Task.Run(() => handleAsyncResultsAsTheyHappen(asyncEnumerable));
        int n = await someOtherTask();
        source.YieldReturn(n);
        var r = await ChildMethod(source);
        source.Complete();  // this call would cause the HandlerTask to complete.
        return r;
    }

    private async Task<string> ChildMethod(AsyncEnumerableSource<int> source) {
        source.YieldReturn(5);
        await SomeOtherCall();
        source.YieldReturn(10);
        return "hello";
    }
}

With the above code, the handleAsyncResultsAsTheyHappen task would see whatever values got passed into YieldReturn. So it would see the n from the above code, as well as the 5 and the 10 from ChildMethod.

Answer 1

You're much better off if you can structure your code to take advantage of yield return and await foreach. E.g., this code does almost the same thing:

public async Task Consume()
{
    var source = ParentMethod();
    HandlerTask = Task.Run(async () => { await foreach (var item in source) { Console.WriteLine(item); } });
}

public async IAsyncEnumerable<int> ParentMethod()
{
    await Task.Yield();
    yield return 13;
    await foreach (var item in ChildMethod())
        yield return item;
}

private async IAsyncEnumerable<int> ChildMethod()
{
    yield return 5;
    await Task.Yield();
    yield return 10;
}

However, if you really need an "async enumerable source", you need to first recognize one thing. TaskCompletionSource<T> holds the results, i.e., the T (or exception). It's acting as a container. The result can be set before the task is awaited. It's the same thing with the "async enumerable source" - you'd need it to be able to hold results before any items are taken from it. The "async enumerable source" would need to hold multiple results - in this case, a collection.

So what you're actually asking for is "a collection that can be consumed as an asynchronous enumerable". There are a few possibilities here, but the one I'd recommend is a Channel:

public async Task<string> ParentMethod()
{
  var source = Channel.CreateUnbounded<int>();
  var sourceWriter = source.Writer;
  IAsyncEnumerable<int> asyncEnumerable = source.Reader.ReadAllAsync();
  HandlerTask = Task.Run(async () => { await foreach (var item in asyncEnumerable) Console.WriteLine(item); });
  await Task.Yield();
  await sourceWriter.WriteAsync(13);
  var r = await ChildMethod(sourceWriter);
  sourceWriter.Complete();
  return r;
}

private async Task<string> ChildMethod(ChannelWriter<int> sourceWriter)
{
  await sourceWriter.WriteAsync(5);
  await Task.Yield();
  await sourceWriter.WriteAsync(10);
  return "hello";
}