Return to Revisions

In a context where thread affinity is required and a SynchronizationContext is not available (as in console applications or windows services), a blocking implementation is required.

I have just a few recommendations:

WaitHandle.WaitAny(new WaitHandle[] {killSubProc, channelWaitHandle});

This wait can cause an issue. If the remote application sends a message and while execution of xxx() continues on the main thread, another message is received, that new message will not be processed until another new message is received.

Main Thread   -> Wait               xxx()----------------- WaitAny       xxx()
Thread X      ->         Message Set
Thread Y      ->                          Message Set                     
Thread Z      ->                                              Message Set
   

Another suggestion:

Move the queue and queueing/dequeueing logic into a class to de-couple it's implementation from invocation of XXX (Make a reusable blocking queue)

Here we have a very generic Message class to store the callback handler and the payload (e.data in your case)

internal class Message
{
    private readonly SendOrPostCallback handler;
    private readonly object payload;

    internal Message(SendOrPostCallback handler, object payload)
    {
        this.handler = handler;
        this.payload = payload;
    }

    internal void Execute()
    {
        handler(payload);
    }
}

And another very short class for keeping track of the queue and to execute the operations (XXX in your case). Please check the use of Queue.Synchronized

public class BlockingMessageSynchronizer
{
    private Queue synchronizationQueue;
    private bool stopRequested;

    public void Start()
    {
        lock (this)
        {
            if (synchronizationQueue != null)
            {
                throw new InvalidOperationException("The synchronization has already started.");
            }
            synchronizationQueue = Queue.Synchronized(new Queue());
            stopRequested = false;
        }

        while (!stopRequested)
        {
            // Thread.Sleep(1);
            ProcessQueue();
        }

        ProcessQueue();
        synchronizationQueue = null;
    }

    public void Post(SendOrPostCallback messageHandler, object payload)
    {
        if (synchronizationQueue == null || stopRequested)
        {
            lock (this)
            {
                if (synchronizationQueue == null || stopRequested)
                {
                    throw new InvalidOperationException("The synhronization is not started or is stopped");
                }
            }
        }
        synchronizationQueue.Enqueue(new Message(messageHandler, payload));
    }

    public void Stop()
    {
        stopRequested = true;
    }

    private void ProcessQueue()
    {
        while (synchronizationQueue.Count > 0)
        {
            ((Message)synchronizationQueue.Dequeue()).Execute();
        }
    }
}

This class can be used on ay thread to force execution of the callback handler on that thread, by blocking that thread. This implementation will cause 100% cpu since there is no Wait operations, you can consider Thread.Sleep(1) before the call to ProcessQueue() in the while loop in Start()

And finally, your stdout or stderr handlers will look like:

process.OutputDataReceived += (sender, e) =>
{
    if (e.Data != null)
    {
        synchronizer.Post(HandleCrossThreadMessage, e.Data);
    }
};

process.ErrorDataReceived += (sender, e) =>
{
    if (e.Data != null)
    {
        synchronizer.Post(HandleCrossThreadMessage, "STDERR");
        synchronizer.Post(HandleCrossThreadMessage, e.Data);
    }
};

process.Exited += (sender, e) =>
{
    synchronizer.Post(HandleCrossThreadMessage, "EXIT");
    synchronizer.Stop();
};

BlockingMessageSynchronizer synchronizer = new BlockingMessageSynchronizer();

process.Start();

process.BeginOutputReadLine();
process.BeginErrorReadLine();

synchronizer.Start();

When you call synchronizer.Stop(); inside the exit event, the remaining messages in the queue will be processed and the execution will get out of the synchronizer.Start(); blocking call.