Is there a way to use System.Diagnostics.Process in an IAsyncEnumerator?

Question

First off, I am targeting .Net Core 3.1 and C#8.

I want something like this.

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();
  }
}

When I try this setup I get an errors from await foreach(string result in RunProcessAsync(args))

CS8417 'Process': type used in an asynchronous using statement must be implicitly convertible to 'System.IAsyncDisposable' or implement a suitable 'DisposeAsync' method.

and this error from yield return e.Data;

CS1621 The yield statement cannot be used inside an anonymous method or lambda expression

The goal is this. I have an exe that does some stuff and writes information to the error output stream (not sure if that's its real name). I want to take in those writes as they are made, parse them for the information I want and store them in an object for later use.

I am a pretty novice coder and very new to asynchronous coding. I tested the functionality of RunProcessAsync but in a synchronous manner; where it was called and just wrote all the raw data to the output window without returning any of it to the calling method. That worked just fine. Also, I have gotten a test asyc stream working with IAsyncEnumerable, but it just used Task.Delay and returned some integers. Now I am trying to combine the to things and my lack of experience is getting in my way.

Thank you for any help you all might give and for helping to increase skill and knowledge of C#.

The first error is pretty clear: you can't use `await using` with Process — Panagiotis Kanavos, Sep 18 '20 at 06:25
You may find this interesting: [Factory for IAsyncEnumerable or IAsyncEnumerator](https://stackoverflow.com/questions/61540896/factory-for-iasyncenumerable-or-iasyncenumerator) — Theodor Zoulias, Sep 18 '20 at 06:55

Peter Duniho · Accepted Answer · 2020-09-18T07:02:53.083

Without a minimal, reproducible example, it will be impossible to address your concern completely. But we can deal with the two specific issues you've raised.

First, if your object (such as Process) doesn't support IAsyncDisposable, then just don't use that. Use the synchronous using statement instead.

As far as the yield return in the method goes, if you take a moment you'll probably see that what you tried to write doesn't make any sense. How would the event handler, which is a completely different method, be able to cause the current method to yield a new value? You need the event handler to signal to the current method when the event occurs. You can do this in a variety of ways, but SemaphoreSlim is one of the more straightforward ways.

Putting those together, you might get something like this:

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;
    }
  }
}

Since you didn't provide an actual MCVE, it's not feasible for me to try to reconstruct your scenario from scratch. So the above isn't compiled, never mind tested. But it should show the gist.

Which is, you need to keep your iterator method asynchronous (which means you can't block on a call to WaitForExit()), and you need to somehow move the data received by the ErrorDataReceived event handler back to the iterator method. In the above, I use a thread-safe queue object in conjunction with a semaphore.

The semaphore count gets increased (via Release()) in the event handler each time a line of data is received, which is then consumed by the iterator method by decreasing the semaphore count (via WaitAsync()) and returning the line received.

There are lots of other mechanisms one could use for the producer/consumer aspect here. There's a well-received Q&A here that discusses async-compatible mechanisms, including a custom version of BlockingCollection<T> that supports async operations, and a mention of the BufferBlock<T> class from TPL Dataflow.

Here is an example that uses BufferBlock<T> (which has semantics very similar to BlockingCollection<T> but includes async handling of the consuming code):

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();
        }
    }
}

Is there a way to use System.Diagnostics.Process in an IAsyncEnumerator?

1 Answers1