如何在 IAsyncEnumerable 发射器函数中正确使用 NpgsqlTransaction？

Question

我不需要捕获异常，但是如果有异常我确实需要回滚：

public async IAsyncEnumerable<Item> Select()
{
    var t = await con.BeginTransactionAsync(token);
    try {
        var batchOfItems = new List<Item>(); //Buffer, so the one connection can be used while enumerating items
        using (var reader = await com.ExecuteReaderAsync(SQL, token)) 
        {
            while (await reader.ReadAsync(token))
            {
                var M = await Materializer(reader, token);
                batchOfItems.Add(M);
            }
        }

        foreach (var item in batchOfItems)
        {
            yield return item;
        }

        await t.CommitAsync();
    }
    catch
    {
        await t.RollbackAsync();
    }
    finally
    {
        await t.DisposeAsync();
    }
}

（此代码是我正在做的简化版本，用于说明目的）

失败并显示以下消息：

不能在带有 catch 子句的 try 块的主体中​​产生值

---

这类似于Yield return from a try/catch block，但它具有新颖的上下文：

• “IAsyncEnumerable”相对较新。
• Postgresql（答案使用内部属性）
• 这个问题有一个更好的标题，明确提到“交易”上下文。具有相同错误消息的其他上下文不会有相同的答案。

这与Why can't yield return appear inside a try block with a catch? 不同。就我而言，上下文更具体：我需要 catch 块来回滚，而不是做任何其他事情。此外，如您所见，我已经知道答案并将其创建为问答组合。从答案中可以看出，该答案与Why can't yield return appear inside a try block with a catch?

无关

Answer 1

如果您可以检查事务是否已提交，您可以将回滚移动到 finally 块，您可以使用 IsCompleted 来完成此操作

public async IAsyncEnumerable<Item> Select()
{
    var t = await con.BeginTransactionAsync(token);
    try {
        var batchOfItems = new List<Item>(); //Buffer, so the one connection can be used while enumerating items
        async using (var reader = await com.ExecuteReaderAsync(SQL, token)) 
        {
            while (await reader.ReadAsync(token))
            {
                var M = await Materializer(reader, token);
                batchOfItems.Add(M);
            }
        }

        foreach (var item in batchOfItems)
        {
            yield return item;
        }

        await t.CommitAsync();
    }
    finally
    {
        if (t.IsCompleted == false) //Implemented on NpgsqlTransaction, but not DbTransaction
            await t.RollbackAsync();
        await t.DisposeAsync();
    }
}

注意：catch 块已被移除，finally 块在开头添加了两行。

同样的方法也适用于没有 IsCompleted 的其他 DbTransaction 实现

见https://stackoverflow.com/a/7245193/887092

Answer 2

使用 C# iterator 创建 IAsyncEnumerable 的替代方法是使用第三方库 AsyncEnumerator (package)。

这个库是在 C# 8 出现之前创建异步枚举的主要资源，它可能仍然有用，因为 AFAIK 它不受本机 yield 的限制。您可以将 lambda 主体中的 try、catch 和 finally 块传递给 AsyncEnumerable 构造函数，并从这些块中的任何一个调用 yield.ReturnAsync 方法。

使用示例：

using Dasync.Collections;

//...

public IAsyncEnumerable<Item> Select()
{
    return new AsyncEnumerable<Item>(async yield => // This yield is a normal argument
    {
        await using var transaction = await con.BeginTransactionAsync(token);
        try
        {
            var batchOfItems = new List<Item>();
            await using (var reader = await com.ExecuteReaderAsync(SQL, token))
            {
                while (await reader.ReadAsync(token))
                {
                    var M = await Materializer(reader, token);
                    batchOfItems.Add(M);
                }
            }
            foreach (var item in batchOfItems)
            {
                await yield.ReturnAsync(item); // Instead of yield return item;
            }
            await transaction.CommitAsync();
        }
        catch (Exception ex)
        {
            await transaction.RollbackAsync();
        }
    });
}

上面例子中的yield不是C#的yield contextual keyword，只是一个同名的参数。如果你愿意，你可以给它另一个名字。

Answer 3

DbTransaction 被认为是管理 SqlConnections 上事务的最佳方式，但 TransactionScope 也是有效的，并且可能在相关场景中帮助其他人

public async IAsyncEnumerable<Item> Select()
{
    using (var scope = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
    {
        con.EnlistTransaction(Transaction.Current); //it's better to open the connection here, then dispose, but this will work
        com = con.CreateCommand(); //Probably need a new command object so it has the transaction context
        var batchOfItems = new List<Item>(); //Buffer, so the one connection can be used while enumerating items
        
        async using (var reader = await com.ExecuteReaderAsync(SQL, token)) 
        {
            while (await reader.ReadAsync(token))
            {
                var M = await Materializer(reader, token);
                batchOfItems.Add(M);
            }
        }

        foreach (var item in batchOfItems)
        {
            yield return item;
        }

        scope.Complete(); //Asynch option not available
        //No need to have explicit rollback call, instead it's standard for that to happen upon disposal if not completed
    }
}