.NET pregunta sobre operaciones de socket asíncronas y estructura de mensaje

Question

He estado buscando ejemplos en todo el mundo sobre cómo tratar el encuadre de mensajes TCP. Veo muchos ejemplos en los que NetworkStream pasa a un objeto StreamReader o StreamWriter y luego utiliza los métodos ReadLine o WriteLine para los mensajes delimitados '\ n'. Mi protocolo de aplicación contiene mensajes que terminan en '\ n', así que NetworkStream parece ser el camino a seguir. Sin embargo, no puedo encontrar ejemplos específicos sobre la forma correcta de manejar todo esto en combinación con sockets asíncronos. Cuando se llama a ReceiveCallback() a continuación, ¿cómo implemento las clases NetworkStream y StreamReader para tratar el encuadre de mensajes? Según lo que he leído, puedo obtener parte de un mensaje en una recepción y el resto del mensaje (incluido el '\ n') en la próxima recepción. ¿Esto significa que podría obtener el final de un mensaje y parte del siguiente? Sin duda, debe haber una manera más fácil de manejar esto.

Tengo el siguiente código:

private void StartRead(Socket socket) 
    { 
     try 
     { 
      StateObject state = new StateObject(); 
      state.AsyncSocket = socket; 

      socket.BeginReceive(state.Buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReceiveCallback), state); 
     } 
     catch (SocketException) 
     { 
      m_Socket.Shutdown(SocketShutdown.Both); 
      Disconnect(); 
     } 
    } 

    private void ReceiveCallback(IAsyncResult ar) 
    { 
     try 
     { 
      StateObject state = (StateObject)ar.AsyncState; 

      int bytes_read = state.AsyncSocket.EndReceive(ar); 

      char[] chars = new char[bytes_read + 1]; 
      System.Text.Decoder decoder = System.Text.Encoding.UTF8.GetDecoder(); 
      int charLength = decoder.GetChars(state.Buffer, 0, bytes_read, chars, 0); 

      String data = new String(chars); 

      ParseMessage(data); 

      StartRead(state.AsyncSocket); 
     } 
     catch (SocketException) 
     { 
      m_Socket.Shutdown(SocketShutdown.Both); 
      Disconnect(); 
     } 
    } 

Answer 1

Básicamente se crea un búfer, y cada vez que recibe datos, añadir los datos a la memoria intermedia y determinar si ya ha recibido uno o más mensajes completos.

Entre ReceiveCallback y StartRead no recibirá ningún mensaje asíncrono (los datos entrantes se almacenarán automáticamente en el nivel de socket) por lo que es el lugar ideal para buscar mensajes completos y eliminarlos del búfer.

Todas las variaciones son posibles, incluida la recepción del final del mensaje 1, más el mensaje 2, más el comienzo del mensaje 3, todo en un solo fragmento.

No recomiendo UTF8-decodificación del fragmento, ya que un carácter UTF8 puede constar de dos bytes, y si se dividen entre fragmentos sus datos podrían estar dañados. Puede mantener un byte [] - búfer (MemoryStream?) Y dividir mensajes en el byte 0x0A en ese caso.

Answer 2

Prefijar los trozos con una longitud es mejor que usar un carácter de separación. No tiene que lidiar con ningún tipo de escape para enviar datos con una nueva línea de esa manera.

Esta respuesta podría no ser importante para usted ahora, porque utiliza características del AsyncCTP, que solo estarán en la próxima versión de .net. Sin embargo, hace las cosas mucho más concisas. Esencialmente escribes exactamente el código que harías para el caso síncrono, pero insertas las declaraciones 'en espera' donde hay llamadas asíncronas.

public static async Task<Byte[]> ReadChunkAsync(this Stream me) { 
     var size = BitConverter.ToUInt32(await me.ReadExactAsync(4), 0); 
     checked { 
      return await me.ReadExactAsync((int)size); 
     } 
    } 

    public static async Task<Byte[]> ReadExactAsync(this Stream me, int count) { 
     var buf = new byte[count]; 
     var t = 0; 
     while (t < count) { 
      var n = await me.ReadAsync(buf, t, count - t); 
      if (n <= 0) { 
       if (t > 0) throw new IOException("End of stream (fragmented)"); 
       throw new IOException("End of stream"); 
      } 
      t += n; 
     } 
     return buf; 
    } 

    public static void WriteChunk(this Stream me, byte[] buffer, int offset, int count) { 
     me.Write(BitConverter.GetBytes(count), 0, 4); 
     me.Write(buffer, offset, count); 
    } 

Answer 3

OK esto es lo que terminé haciendo. Creé un hilo lector que crea un NetworkStream y un StreamReader basados ​​en la transmisión de la red. Luego uso StreamReader.ReadLine para leer en las líneas de esa manera. Es una llamada sincrónica, pero está en su propio hilo. Parece funcionar mucho mejor. Tuve que implementar esto ya que ese es nuestro protocolo para la aplicación (mensajes delimitados por línea nueva). Sé que otras personas estarán buscando por ahí como el infierno por la respuesta como lo hice del código de lectura relevante en mi clase de cliente aquí:

public class Client 
{ 
    Socket    m_Socket; 

    EventWaitHandle  m_WaitHandle; 
    readonly object  m_Locker; 
    Queue<IEvent>  m_Tasks; 
    Thread    m_Thread; 

    Thread    m_ReadThread; 

    public Client() 
    { 
     m_WaitHandle = new AutoResetEvent(false); 
     m_Locker = new object(); 
     m_Tasks = new Queue<IEvent>(); 

     m_Thread = new Thread(Run); 
     m_Thread.IsBackground = true; 
     m_Thread.Start(); 
    } 

    public void EnqueueTask(IEvent task) 
    { 
     lock (m_Locker) 
     { 
      m_Tasks.Enqueue(task); 
     } 

     m_WaitHandle.Set(); 
    } 

    private void Run() 
    { 
     while (true) 
     { 
      IEvent task = null; 

      lock (m_Locker) 
      { 
       if (m_Tasks.Count > 0) 
       { 
        task = m_Tasks.Dequeue(); 

        if (task == null) 
        { 
         return; 
        } 
       } 
      } 

      if (task != null) 
      { 
       task.DoTask(this); 
      } 
      else 
      { 
       m_WaitHandle.WaitOne(); 
      } 
     } 
    } 

    public void Connect(string hostname, int port) 
    { 
     try 
     { 
      m_Socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); 

      IPAddress[] IPs = Dns.GetHostAddresses(hostname); 

      m_Socket.BeginConnect(IPs, port, new AsyncCallback(ConnectCallback), m_Socket); 
     } 
     catch (SocketException) 
     { 
      m_Socket.Close(); 
      OnConnect(false, "Unable to connect to server."); 
     } 
    } 

    private void ConnectCallback(IAsyncResult ar) 
    { 
     try 
     { 
      Socket socket = (Socket)ar.AsyncState; 

      socket.EndConnect(ar); 

      OnConnect(true, "Successfully connected to server."); 

      m_ReadThread = new Thread(new ThreadStart(this.ReadThread)); 
      m_ReadThread.Name = "Read Thread"; 
      m_ReadThread.IsBackground = true; 
      m_ReadThread.Start(); 
     } 
     catch (SocketException) 
     { 
      m_Socket.Close(); 
      OnConnect(false, "Unable to connect to server."); 
     } 
    } 

    void ReadThread() 
    { 
     NetworkStream networkStream = new NetworkStream(m_Socket); 
     StreamReader reader = new StreamReader(networkStream); 

     while (true) 
     { 
      try 
      { 
       String message = reader.ReadLine(); 

       // To keep the code thread-safe, enqueue a task in the CLient class thread to parse the message received. 
       EnqueueTask(new ServerMessageEvent(message)); 
      } 
      catch (IOException) 
      { 
       // The code will reach here if the server disconnects from the client. Make sure to cleanly shutdown... 
       Disconnect(); 
       break; 
      } 
     } 
    } 

    ... Code for sending/parsing the message in the Client class thread. 
}