有没有办法在IAsyncEnumerator中使用System.Diagnostics.Process？

问题描述

首先，我的目标是.Net Core 3.1和C＃8。

我想要这样的东西。

public static async Task<MyDataObj> GetData()
{
  var dataObj = new MyDataObj();
  var args = ArgsHelperFunction(...);
  
  await foreach(string result in RunProcessAsync(args))
  {
     // Process result and store it in dataObj
  }

  return dataObj;
}

private static async IAsyncEnumerable<string> RunProcessAsync(string args)
{
  await using (var myProcess = new Process())
  {
    myProcess.StartInfo.FileName = @"path\to\file.exe"
    myProcess.StartInfo.Arguments = args;
    myProcess.StartInfo.UseShellExecute = false;
    myProcess.StartInfo.RedirectStandardError = true;
    myProcess.StartInfo.CreateNowindow = true;

    myProcess.ErrorDataReceived += (s,e) =>
    {
      yield return e.Data;
    }

    myProcess.Start();
    myProcess.BeginErrorReadLine();
    process.WaitforExit();
  }
}

尝试此设置时，我收到await foreach(string result in RunProcessAsync(args))的错误消息

CS8417'Process'：异步using语句中使用的类型必须隐式转换为'System.IAsyncdisposable'或实现适当的'disposeAsync'方法。

以及来自yield return e.Data;

的错误

CS1621 yield语句不能在匿名方法或lambda表达式内使用

目标是这样。我有一个可执行文件的exe，并将信息写入错误输出流（不确定这是否是它的真实名称）。我想在写入时接受这些写入，将它们解析为我想要的信息，并将它们存储在对象中以备后用。

我是一个相当新手的编码人员，并且是异步编码的新手。我以同步方式测试了RunProcessAsync的功能；在哪里被调用，并将所有原始数据写入输出窗口，而没有将任何原始数据返回给调用方法。那很好。另外，我得到了一个与IAsyncEnumerable一起使用的测试asyc流，但是它只使用了Task.Delay并返回了一些整数。现在，我正在尝试将事物与事物结合起来，而我的经验不足正妨碍着我。

感谢大家提供的任何帮助，并帮助您提高了C＃的技能和知识。

解决方法

如果没有minimal,reproducible example，将不可能完全解决您的问题。但是我们可以处理您提出的两个具体问题。

首先，如果您的对象（例如Process）不支持IAsyncDisposable，则不要使用它。请使用同步using语句。

就方法中​​的yield return而言，花一点时间，您可能会发现尝试编写的内容没有任何意义。事件处理程序（完全不同的方法）如何使当前方法产生新值？您需要事件处理程序在事件发生时向当前方法发出信号。您可以通过多种方式执行此操作，但是SemaphoreSlim是更直接的方式之一。

将它们放在一起，您可能会得到以下信息：

private static async IAsyncEnumerable<string> RunProcessAsync(string args)
{
  using (var myProcess = new Process())
  {
    myProcess.StartInfo.FileName = @"path\to\file.exe";
    myProcess.StartInfo.Arguments = args;
    myProcess.StartInfo.UseShellExecute = false;
    myProcess.StartInfo.RedirectStandardError = true;
    myProcess.StartInfo.CreateNoWindow = true;

    ConcurrentQueue<string> dataQueue = new ConcurrentQueue<string>();
    SemaphoreSlim dataSemaphore = new SemaphoreSlim(0);

    myProcess.ErrorDataReceived += (s,e) =>
    {
      dataQueue.Enqueue(e.Data);
      dataSemaphore.Release();
    }

    myProcess.Start();
    myProcess.BeginErrorReadLine();

    while (true)
    {
      await dataSemaphore.WaitAsync();

      // Only one consumer,so this will always succeed
      dataQueue.TryDequeue(out string data);

      if (data == null) break;

      yield return data;
    }
  }
}

由于您没有提供实际的MCVE，因此我尝试从头开始重建方案是不可行的。因此，以上内容未经编译，无需经过测试。但这应该显示出要点。

这是，您需要使迭代器方法保持异步（这意味着您无法阻止对WaitForExit()的调用），并且需要以某种方式移动ErrorDataReceived事件接收到的数据处理程序返回到迭代器方法。在上面，我结合使用了线程安全队列对象和信号灯。

每次接收到一行数据时，事件处理程序中的信号量计数就会增加（通过Release()），然后由迭代器方法通过减少信号量计数（通过WaitAsync()）来使用并返回收到的行。

在这里，生产者/消费者方面可以使用许多其他机制。 a well-received Q&A here讨论了异步兼容的机制，包括支持异步操作的BlockingCollection<T>的自定义版本，以及TPL Dataflow中提到的the BufferBlock<T> class。

下面是一个使用BufferBlock<T>的示例（其语义与BlockingCollection<T>非常相似，但包括对消费代码的异步处理）：

static async IAsyncEnumerable<string> RunProcessAsync(string args)
{
    using (var process = new Process())
    {
        myProcess.StartInfo.FileName = @"path\to\file.exe";
        process.StartInfo.Arguments = args;
        process.StartInfo.UseShellExecute = false;
        process.StartInfo.RedirectStandardError = true;
        process.StartInfo.CreateNoWindow = true;

        BufferBlock<string> dataBuffer = new BufferBlock<string>();

        process.ErrorDataReceived += (s,e) =>
        {
            if (e.Data != null)
            {
                dataBuffer.Post(e.Data);
            }
            else
            {
                dataBuffer.Complete();
            }
        };

        process.Start();
        process.BeginErrorReadLine();

        while (await dataBuffer.OutputAvailableAsync())
        {
            yield return dataBuffer.Receive();
        }
    }
}

,

像这样吗？

using CliWrap;
using CliWrap.EventStream;

var cmd = Cli.Wrap("foo").WithArguments("bar");

await foreach (var cmdEvent in cmd.ListenAsync())
{
    switch (cmdEvent)
    {
        case StartedCommandEvent started:
            _output.WriteLine($"Process started; ID: {started.ProcessId}");
            break;
        case StandardOutputCommandEvent stdOut:
            _output.WriteLine($"Out> {stdOut.Text}");
            break;
        case StandardErrorCommandEvent stdErr:
            _output.WriteLine($"Err> {stdErr.Text}");
            break;
        case ExitedCommandEvent exited:
            _output.WriteLine($"Process exited; Code: {exited.ExitCode}");
            break;
    }
}

在CliWrap中可用。