使用 TaskCompletionSource 构建 IAsyncEnumerable

Question

我有一个接受IEnumerable 并使用yield 运算符转换的方法返回它。要转换可枚举的一个元素，我首先需要知道另一个元素的值。因此，我想到了使用TaskCompletionSource 来创建类似“承诺”的东西。

这里的问题是，如果第一个TestFieldA 的值不是“a”，那么这段代码会导致死锁。一种解决方案是在将枚举传递给方法之前对其进行排序——在这种情况下，完全不需要TaskCompletionSource。但是，我想知道是否可以在没有这个的情况下完成。我也知道这可以通过一些 LINQ 查询来完成，但这需要多次枚举输入，我想避免这种情况。

这就是我想要实现的目标。 （仅适用于第一个 TestFieldA == "a"）

class Test
{
    public string TestFieldA {get;set;}
    public int TestFieldB {get;set;}
}


private async IAsyncEnumerable<Test> Transform(IEnumerable<Test> inputEnumerable)
{
    var tcs = new TaskCompletionSource<int>();

    foreach(var input in inputEnumerable)
    {
        if (input.TestFieldA == "a")
        {
            tcs.SetResult(input.TestFieldB);
            yield return input;
        }
        else
        {
            input.TestFieldB -= await tcs.Task;
            yield return input;
        }
    }
}

Answer 1

您目前的计划似乎取决于能否及时返回。我建议只将不合适的项目存储在队列中（而不是让它们产生），直到找到具有 TestFieldA 值的合适项目。

此时，您将所有排队的项目出列，使用现在找到的值并依次生成每个项目。然后生成具有所需 TestFieldA 值的项目。

你如何从那里开始有点不清楚，因为我不知道如果 a) 找到另一个 a 项目并且 b) 如果没有找到 a 项目该怎么办。

这里不需要Task(CompletionSource)、async 或IAsyncEnumerable - 在找到a 值之前，你无法产生任何东西 - 除非你可以使用时间机器。

---

还要记住，迭代器依赖于让他们的调用者反复请求新项目以取得进展 - 你在每个 yield 处暂停，直到他们这样做。因此，如果生成的项目有任何关于它们的“Task-like”，那么考虑尽早尝试yield 项目将是极端冒险的；调用者可能会决定await 一个而不是继续您需要的枚举。

Answer 2

一个想法可能是返回一个可枚举的任务，而不是一个IAsyncEnumerable。像这样的：

private IEnumerable<Task<Test>> Transform(IEnumerable<Test> source)
{
    var tcs = new TaskCompletionSource<int>(
        TaskCreationOptions.RunContinuationsAsynchronously);

    foreach (var item in source)
    {
        if (item.TestFieldA == "a")
        {
            tcs.TrySetResult(item.TestFieldB);
        }
        yield return TransformItemAsync(item);
    }

    async Task<Test> TransformItemAsync(Test input)
    {
        var value = await tcs.Task.ConfigureAwait(false);
        input.TestFieldB -= value;
        return input;
    }
}

如果调用者试图按顺序等待每个任务，这仍然会产生死锁问题。为了解决这个问题，调用者应该有办法以某种方式等待任务完成。 Stephen Cleary 的Nito.AsyncEx 库中有类似的东西，扩展方法OrderByCompletion：

// Creates a new collection of tasks that complete in order.
public static List<Task<T>> OrderByCompletion<T>(this IEnumerable<Task<T>> @this);

如果需要，您也可以从here 获取源代码。