¿Cómo se implementa un oyente de tuberías con nombre reutilizable que se ejecuta de forma asíncrona?

Question

No encuentro un buen ejemplo de cómo crear un receptor de canalizaciones con nombre reutilizable que se ejecute de forma asíncrona. Puedo hacer un oyente reutilizable:

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

    while (true) 
    { 
      pipeServer.WaitForConnection(); 

      StreamReader reader = new StreamReader(pipeServer); 

      MessageBox.Show(reader.ReadLine()); 

      pipeServer.Disconnect(); 
    } 

y puedo hacer un oyente asíncrono:

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

Pero me parece que no puede obtener tanto ir. ¿Hay un buen ejemplo para esto? También me preocupan los mensajes parcialmente enviados, ya que creo que es un problema con las comunicaciones asíncronas como esta.

Actualización: Estoy un poco más cerca.

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

que leerá con éxito una vez, y continuará bucle, con ReadLine devolver una cadena vacía vacía después de la lectura con éxito inicial. Entonces claramente no está bloqueando, y está intentando leer de nuevo. El problema es que si envío el mismo mensaje por segunda vez, no aparece, y mi guionista dice que está recibiendo el error 2316 (aunque no entiendo lo que eso significa). Creo que solo tengo que hacer algo similar a esto, donde la tubería se limpia cada vez, como la primera muestra de código que enumeré, pero todavía no he conseguido que funcione.

Answer 1

creo que ya lo tengo:

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

Esto aceptará múltiples mensajes como vienen, y se cerrará tan pronto como correr se establece en false (en otro hilo, obviamente). Parece ser lo que necesito. ¿Alguien puede verificar que no estoy haciendo nada tonto?

Answer 2

I'm also concerned about partially sent messages

Ellos no son un problema con NamedPipes nativa utilizando el API (Win32), por lo que dudo mucho que son un problema utilizando .NET. Sin embargo en el native documentation sí dice:

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.

(Nota ERROR_MORE_DATA es 234.)

La documentación también se dice, por bandera FILE_FLAG_OVERLAPPED (el equivalente natural de 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 Siempre he usado operaciones de IO asíncronas con canalizaciones con nombre asíncronas (es decir, Stream.BeginRead), pero esto significa perder la funcionalidad de TextReader, pero luego PipeTransmissionMode.Message se define en términos de transmisión de grupos de bytes de todos modos.