问题标签 [iasyncenumerable]

我正在尝试为IAsyncEnumerable<T>集合创建一个自定义 DataReader，以便通过SqlBulkCopy. 我正在按照这个问题中概述的代码示例和解决方案的行 -如何使用 SqlBulkCopy 编写 IAsyncEnumerable

这是我的 DataReader 的要点：

我的问题是，当我将 datareader 传递给SqlBulkCopy.WriteToServerAsync(reader)方法时，在第一次调用DataReader.Read()它时会引发以下错误：

System.InvalidOperationException: 'Operation is not valid due to the current state of the object.'

有谁知道我的代码哪里出错了？

我们正在为IAsyncEnumerable. 我们的最终目标是能够使用System.Linq.Async. 不幸的是，当前的命名空间冲突使我们无法这样做。与此同时，我们正在编写自己的代码并尝试尽可能地符合他们的签名，以使换入/换出尽可能容易。

我们注意到一种模式，但不知道为什么它是一件事或它是否重要。考虑以下两个函数。

和

System.Linq.Async在大多数/所有扩展中使用前者。我们不知道为什么会这样。为什么不像前者那样简单地写呢？最终，这两者之间有什么区别？一个似乎只是简单地包装另一个以async从签名中删除。我们使用 C#9 作为上下文。我们应该使用哪个，为什么？

在我的 .Net Core 应用程序中，第 3 方库中有一个写入System.IO.Stream接口的方法（它将流接口作为参数并写入它），但我希望该数据进入我的数据源，该数据源期望数据作为IAsyncEnumerable<bytes>溪流。我着手编写实现Stream接口的代码，因此当Write()被调用时它会写入IAsyncEnumerable<bytes>，然后认为“这必须在之前完成” - 似乎它是通用的。

那么在第 3 方库中是否有标准实现，或者我缺少任何“巧妙的技巧”？

我想知道为什么Observable当我不等待它的来源时我没有产生任何元素IAsyncEnumerable。我正在使用 a Selectover the Observable，并且没有任何结果，我的代码没有执行。

在上面的例子中，代码工作并被调用，而下面没有任何输出。

我是否需要在某处存储对这个 observable 的引用，或者应该如何做到这一点？

我正在使用第三方库，它充当发布-订阅消息代理的接口。经纪人是 Solace PubSub+。

对于订阅者，供应商库采用“通过回调推送消息”模式。

我正在围绕供应商库编写自己的包装库，以使其他开发人员更容易使用（隐藏库如何与网络通信的所有内部结构等等）。

同样，我认为将订阅者提要公开IAsyncEnumerable为System.Threading.Channels. 我有两个担忧：

1. 这里的渠道合适吗，还是我过度设计了这个？即，是否有更“C# 惯用”的方式来包装回调？
2. 我的EnumerableBroker包装器实现是安全的，还是我在某个地方陷入了异步陷阱？

我意识到第一个问题可能比 SO 更适合 CodeReview，但由于该问题的答案也与第二个问题有关，因此将它们放在一起似乎是合适的。值得注意的是：我正在避免IObservable/Rx，因为我的目标是让我的界面比供应商的更基本，而不是要求其他开发人员和我自己学习 Rx！了解生产者和消费者进程是如何独立的对于中间的通道也是微不足道的，而对于一个可观察的，我的第一个心理过程是“好的，生产者和消费者仍然独立吗？乍一看，我必须现在学习调度程序......天哪，我只使用一个await foreach怎么样？”

这是一个没有 的消费消息的最小模型EnumerableBroker：

现在有了最小的复制EnumerableBroker（仍然使用上面列出的相同的模拟Broker类）。这里至少有一个好处似乎是，如果订阅者需要做很多工作来处理消息，它不会占用代理的线程 - 至少在缓冲区填满之前。这似乎可以正常工作，但我已经学会警惕我对异步的有限掌握。

我想将 IAsyncEnumerable 用作 akka 流的 Source。但是我没找到，怎么办。

此代码的 Source 类中没有合适的方法。

如何将 IAsyncEnumerbale 用于 Source？

所以我试图采用一种IAsyncEnumerable方法并在 PowerShell 中传递所述方法的结果。我知道 PowerShell 没有异步支持，通常，人们使用它Task.GetAwaiter().GetResult()作为获得结果的手段。

但是，这种方法对方法不起作用（或者至少我不知道如何实现它）IAsyncEnumerable。

我的具体情况要复杂一些，但让我们举个例子：

这个 cmdlet 当然是一个虚拟的，只接受一个整数和“狗”、“猫”或“马”，但我更关心的是如何InternalMethodAsync()在 Cmdlet 中处理。挑战在于绕过IAsyncEnumerable.

我正在构建一个客户端，用于持续使用数据源中的新记录。集成是基于拉的，我的客户会定期查询数据源以获取新记录。我将IAsyncEnumerable其用作此连续新记录流的返回类型。以下是相关方法的要点：

这是对 IAsyncEnumerable 的适当使用吗？这个流应该是“永无止境的”或“连续的”（至少在 cancelToken 或错误之前）。

我有几种方法都结束了：

例如：

我想删除这个重复。

如果我尝试将其重构为：

它不会编译，因为我无法从迭代器返回值。

如果我尝试 return yield return YieldAppointments(cursor);，它将无法编译，因为：

严重性代码描述项目文件行抑制状态错误 CS0266 无法将类型“System.Collections.Generic.IAsyncEnumerable<DataAccessLayer.Entities.Praxedo.FieldServiceAppointment>”隐式转换为“DataAccessLayer.Entities.Praxedo.FieldServiceAppointment”。存在显式转换（您是否缺少演员表？） DataAccessLayer C:\projects\EnpalPraxedoIntegration\DataAccessLayer\DbServices\FieldServiceAutomationDbService.cs 78 Active

所以我试图

yield return (IAsyncEnumerable<FieldServiceAppointment>) YieldAppointments(cursor);

和

yield return YieldAppointments(cursor) as IAsyncEnumerable <FieldServiceAppointment>;

其中任何一个都会产生以下编译器错误：

严重性代码描述项目文件行抑制状态错误 CS0266 无法将类型“System.Collections.Generic.IAsyncEnumerable<DataAccessLayer.Entities.Praxedo.FieldServiceAppointment>”隐式转换为“DataAccessLayer.Entities.Praxedo.FieldServiceAppointment”。存在显式转换（您是否缺少演员表？） DataAccessLayer C:\projects\EnpalPraxedoIntegration\DataAccessLayer\DbServices\FieldServiceAutomationDbService.cs 78 Active

所以我尝试了：

但这不会编译，因为

严重性代码描述项目文件行抑制状态错误 CS1624“FieldServiceAutomationDbService.YieldAppointments(IAsyncCursor)”的主体不能是迭代器块，因为“任务”不是迭代器接口类型 DataAccessLayer C:\projects\EnpalPraxedoIntegration\DataAccessLayer\DbServices\FieldServiceAutomationDbService。 cs 81 主动

有没有办法使这项工作？

为什么会UseEventsFail在下面的代码中抛出？难道是我在不等待最后一个MoveNextAsync()任务的情况下处理了异步枚举器？这个例子是我真实程序的简化复制，所以我需要处理异步枚举器来释放它的资源。并且Task.CompletedTask通常Task.Delay()用作UseEvents(). 如果枚举器任务在超时任务之前完成，则不会引发异常。

异常的堆栈跟踪：

at Program.<<<Main>$>g__GenerateEvents|0_3>d.System.IAsyncDisposable.DisposeAsync()

代码：