嵌套IAsyncEnumerable

Question

我正在努力解决传递给集成的问题。到目前为止，最好的方法是返回IAsyncEnumerable<IAsyncEnumerable<T>>而不是IAsyncEnumerable<T>，然后将其扁平化。我对此并不兴奋，所以我在这里寻求建议。

其基本目标是允许一个人从子方法中获取一部分IAsyncEnumerable，而不必发疯。下面的Counting()方法是我的尝试。

我认为设计递归对象以允许更深层次的嵌套不会有太大的麻烦。但我还是不太喜欢。

public static class AsyncEnumerables
{
    public static async IAsyncEnumerable<T> Flatten<T>(IAsyncEnumerable<IAsyncEnumerable<T>> enumerables)
    {
        await foreach (var e in enumerables)
        {
            await foreach (var t in e)
            {
                yield return t;
            }
        }
    }

    public static async IAsyncEnumerable<T> Singleton<T>(T t)
    {
        await Task.CompletedTask;
        yield return t;
    }
}

public class AsyncEnumerableTests
{
    public async IAsyncEnumerable<int> Children()
    {
        yield return 1;
        await Task.Delay(1000);
        yield return 2;
    }

    public async IAsyncEnumerable<IAsyncEnumerable<int>> Counting()
    {
        yield return AsyncEnumerables.Singleton(0);
        await Task.Delay(100);
        yield return Children();
        yield return AsyncEnumerables.Singleton(3);
    }

    [Fact]
    public async Task Counting_Counts()
    {
        var numbers = new List<int>();
        await foreach (var n in AsyncEnumerables.Flatten(Counting())) {
            numbers.Add(n);
        }
        Assert.Equal(4, numbers.Count);
        for (int i=0; i<4; i++)
        {
            Assert.Equal(i, numbers[i]);
        }
    }
}

Answer 1

在设计API时，有几件事值得考虑，API应该返回IAsyncEnumerable<IAsyncEnumerable<T>>或返回IAsyncEnumerable<T>。

IAsyncEnumerable < IAsyncEnumerable >

Pros

• 从消费者的角度来看，它是灵活的。它允许您使用单个或多个消费者模型。
• 在多个使用者的情况下，吞吐量可能更高，因为并行处理潜力
• 它允许对不同的提供商使用不同的Timeouts。
• 它允许在两个yield return语句之间进行自定义代码注入。

Cons

• 在单一消费者的情况下，摄入逻辑可能要复杂一些(与扁平版本相比)。
• 嵌套异步循环会使异常处理逻辑(以及调试)复杂化。

IAsyncEnumerable

Pros

• 对于单个消费者来说，使用起来更容易。
• 如果多提供者模型只是一个实现细节，那么它可以将这些信息隐藏在一个很好的抽象之后。
• 它提供了更容易处理错误的功能。

Cons

• 更难支持多个消费者模型。
• 很难在yield returns之间插入自定义代码。
• 跨越基于不同提供者的新处理是不可能的。

如果您正在考虑提供一个扁平的API，那么我建议检查这个包，称为System.Interactive.Async。在这个包中有一个名为AsyncEnumerableEx的类，它定义了一个合并函数。