如何将 Task<IEnumerable<T>> 转换为 IAsyncEnumerable<T>？

Question

我正在逐步将Ix.NET引入一个旧项目中。 我有许多存储级别的API返回Task<IEnumerable<T>>，但是我想将它们适应于IAsyncEnumerable<T>以便在整个系统中使用。 看起来应该有一个辅助方法（例如对于IEnumerable的.ToAsyncEnumerable()）来帮助解决这个问题，但我找不到任何东西... 我必须实现自己的自定义枚举器吗？ （不难，但我不想重复造轮子）

- Kevin Halverson

---

你为什么更喜欢使用 IAsyncEnumerable<T> 而不是 IObservable<T>？ - Enigmativity

在服务器端，我想使用 IAsyncEnumerable<T> 以便与 C# 8.0 异步流保持一致（我最终想采用这种方案）。在客户端，我将把由此可枚举对象产生的响应流包装在一个可观察对象中。 - Kevin Halverson

Answer 1

Task<IEnumerable<T>> GetSomeResults<T>()
{
    throw new NotImplementedException();
}

async IAsyncEnumerable<T> GetAsyncEnumerable<T>()
{
    var results = await GetSomeResults<T>();
    foreach(var item in results)
    {
        yield return item;
    }
}

- liubiaoyong

Answer 2

根据Theodor Zoulias的评论，System.Linq.Async是一个来自.NET Foundation的NuGet包，支持ToAsyncEnumerable()。

示例用法：

    var tasks = new Task[0]; // get your IEnumerable<Task>s
    tasks.ToAsyncEnumerable();

- Brett

Answer 3

我也在寻找同样的东西，根据这里的回复，我认为确实没有像AsAsyncEnumerable()这样的方法。所以这是我最终做的事情，希望能对其他人有所帮助。

public static class AsyncEnumerableExtensions {
    public struct AsyncEnumerable<T> : IAsyncEnumerable<T> {
        private readonly IEnumerable<T> enumerable;

        public AsyncEnumerable(IEnumerable<T> enumerable) {
            this.enumerable = enumerable;
        }

        public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = new CancellationToken()) {
            return new AsyncEnumerator<T>(enumerable?.GetEnumerator());
        }
    }

    public struct AsyncEnumerator<T> : IAsyncEnumerator<T> {
        private readonly IEnumerator<T> enumerator;

        public AsyncEnumerator(IEnumerator<T> enumerator) {
            this.enumerator = enumerator;
        }

        public ValueTask DisposeAsync() {
            enumerator?.Dispose();
            return default;
        }

        public ValueTask<bool> MoveNextAsync() {
            return new ValueTask<bool>(enumerator == null ? false : enumerator.MoveNext());
        }

        public T Current => enumerator.Current;
    }

    public static AsyncEnumerable<T> AsAsyncEnumerable<T>(this IEnumerable<T> that) {
        return new AsyncEnumerable<T>(that);
    }

    public static AsyncEnumerator<T> AsAsyncEnumerator<T>(this IEnumerator<T> that) {
        return new AsyncEnumerator<T>(that);
    }
}

- Lucero

---

实际上，像AsAsyncEnumerable这样的方法已经存在于dotnetfoundation拥有的System.Linq.Async包中，并且它被命名为ToAsyncEnumerable。 - Theodor Zoulias

Answer 4

public static async IAsyncEnumerable<T> ToAsyncEnumerable<T>(this IEnumerable<T> enumerable)
{
    using IEnumerator<T> enumerator = enumerable.GetEnumerator();

    while (await Task.Run(enumerator.MoveNext).ConfigureAwait(false))
    {
        yield return enumerator.Current;
    }
}

- Ahmed Hesham

Answer 5

如果你谈论的是Web API，Task<IEnumerable<T>> 是一种异步生成 IEnumerable<T> 的方法。

不管这个 IEnumerable<T> 是同步还是异步生成的，整个列表将被作为HTTP响应发送。

如果客户端调用某种流式传输或对服务器发起多个请求以获取唯一结果列表（分页），则可以利用 IAsyncEnumerable<T>。

- Paulo Morgado

---

是的，没错！ :-) 对于服务器端，我想要启用可枚举对象的异步消费，而在客户端，我想要使用流式响应。 - Kevin Halverson

你打算如何做到这一点？ - Paulo Morgado

在服务器端，我可以使用 System.Interactive.Async 中的 Linq 扩展方法自然地处理 IAsyncEnumerable。对于客户端，我将构建一个可观察对象，该对象使用分块传输编码（例如 https://medium.com/@deaniusaur/how-to-stream-json-data-over-rest-with-observables-80e0571821d3）对响应进行处理。 - Kevin Halverson

我肯定在这里漏掉了什么。IAsyncEnumerable<T> 用于从源异步拉取数据。我没有看到在 WebAPI 上这样做的直接优势。推送到客户端的是 IEnumerable<T>。也许如果您能展示一些代码... - Paulo Morgado

所以，我想在客户端实现的是异步地通过开放式GET请求流式传输实体（类型为T），同时从服务器上异步获取它们。话虽如此，这一切都偏离了最初的问题。把客户端部分放在一边，只看服务器部分，我想能够从Task<IEnumerable<T>>创建一个IAsyncEnumerable<T>，我只是在寻找最优雅的方法来实现它（而不编写自己的枚举器）。 - Kevin Halverson

即，我需要实现以下接口：interface IDataSource<T> { IAsyncEnumerable<T> GetData(); } ...而我用于获取数据的客户端具有以下签名：class OldDatabaseClient<T> { Task<IEnumerable<T>> GetData() } 最简单（相对高效）的实现接口的方法是什么？ - Kevin Halverson