How do you implement a reusable named pipe listener that runs asynchronously?

Question

I can't find a good example of how to create a reusable named pipe listener that runs asynchronously. I can make a reusable listener:

NamedPipeServerStream pipeServer = new NamedPipeServerStream("MyPipe", PipeDirection.InOut);

    while (true)
    {
            pipeServer.WaitForConnection();

            StreamReader reader = new StreamReader(pipeServer);

            MessageBox.Show(reader.ReadLine());

            pipeServer.Disconnect();
    }

and I can make an asychronous listener:

NamedPipeServerStream pipeServer = new NamedPipeServerStream("MyPipe", PipeDirection.InOut, 1, PipeTransmissionMode.Message, PipeOptions.Asynchronous);

    pipeServer.BeginWaitForConnection((a) =>
    {
        pipeServer.EndWaitForConnection(a);

        StreamReader reader = new StreamReader(pipeServer);
        MessageBox.Show(reader.ReadLine());

    }, null);

But I can't seem to get both going. Is there a good example for this? I'm also concerned about partially sent messages, as I believe that is an issue with asynchronous communications like this.

Update: I'm a little closer.

pipeServer = new NamedPipeServerStream("MyPipe", PipeDirection.InOut, 1, PipeTransmissionMode.Message, PipeOptions.Asynchronous);

pipeServer.BeginWaitForConnection((a) =>
{
    pipeServer.EndWaitForConnection(a);

    StreamReader reader = new StreamReader(pipeServer);

    while (running)
    {
        String text = reader.ReadLine();

        if (String.IsNullOrEmpty(text) == false)
        {
            MessageBox.Show(text);
        }
    }

    MessageBox.Show("Done!");

}, null);

That will read successfully once, and will continue looping, with ReadLine returning an empty empty string after the initial successful read. So it's clearly not blocking, and is attempting to read again. The problem is if I send the same message a second time, it doesn't get picked up, and my pipe writer says it's receiving error 2316 (though I can't figure out what that means). I think I just need to do something similar to this where the pipe gets cleaned up each time, like the first code sample I listed, but I haven't gotten that to work yet.

Answer 1

I think I've got it:

pipeServer = new NamedPipeServerStream("MyPipe", PipeDirection.InOut, 1, PipeTransmissionMode.Message, PipeOptions.Asynchronous);

Boolean connectedOrWaiting = false;

Byte[] buffer = new Byte[65535];

while (running)
{
    if (!connectedOrWaiting)
    {                   
        pipeServer.BeginWaitForConnection((a) => { pipeServer.EndWaitForConnection(a); }, null);

        connectedOrWaiting = true;
    }

    if (pipeServer.IsConnected)
    {
        Int32 count = pipeServer.Read(buffer, 0, 65535);

        if (count > 0)
        {
            UTF8Encoding encoding = new UTF8Encoding();
            String message = encoding.GetString(buffer, 0, count);

            MessageBox.Show(message);
        }

        pipeServer.Disconnect();

        connectedOrWaiting = false;
    }
}

This will accept multiple message as they come, and will shut down as soon as running is set to false (in another thread, obviously). It appears to be what I need. Can someone verify that I'm not doing anything silly?

Answer 2

I'm also concerned about partially sent messages

They are not an issue with NamedPipes using the native (Win32) APIs, so I very much doubt they are a problem using .NET. However in the native documentation it does say:

Data is written to the pipe as a stream of messages. The pipe treats the bytes written during each write operation as a message unit. The GetLastError function returns ERROR_MORE_DATA when a message is not read completely. This mode can be used with either PIPE_READMODE_MESSAGE or PIPE_READMODE_BYTE.

(Note ERROR_MORE_DATA is 234.)

The documentation also says, for flag FILE_FLAG_OVERLAPPED (the native equivalent of PipeOptions.Asynchronous):

Overlapped mode is enabled. If this mode is enabled, functions performing read, write, and connect operations that may take a significant time to be completed can return immediately.

I've always used asynchronous IO operations with asynchronous named pipes (i.e. Stream.BeginRead), but this does mean losing the functionality of a TextReader, but then PipeTransmissionMode.Message is defined in terms of transmitting groups of bytes anyway.