TCP client reading socket asynchronously

Question

I'm relatively new to .Net/UWP networking and have been trying to make sense of the APIs. I'd like to implement a TCP client that is capable of reading and sending messages independently (there is no protocol and all messages are independent).

Typically (e.g. in a POSIX C++ server), this would be done by spinning up a thread and reading continuously (sleeping periodically if no data) or using asynchronous I/O (epoll, select-poll, etc.)

I've implemented a UWP-compatible client that attempts to use ReadAsync as well as cancellation tokens to shut the client down.

This is a pretty rough cut and I'd like to flesh it out more, especially wrt error handling. Some questions:

1. Is looping continuously as I've done in ReadMessages() the recommended/idiomatic way of using these functions?

2. Am I using the cancellation tokens appropriately? My intent is to be able to kill a ReadAsync() request when the client object is destructed (or, eventually, when a Disconnect() method is added).

3. Where should I be detecting disconnects? try-catch blocks around the ReadAsync methods?

Lastly, messages (their byte[] buffers, rather), would be output using a fast queue. I've commented that portion out. The intent is to use this as a component in a larger application (e.g., Unity).

namespace Messages
{
  [StructLayout(LayoutKind.Sequential, Pack = 1)]
  public struct Header
  {
    public UInt32 size;
    public UInt32 id;
    public UInt64 received_timestamp;
  }

  [StructLayout(LayoutKind.Sequential, Pack = 1)]
  public struct PingMessage
  {
    public Header header;
    public UInt64 timestamp;

    public void Init()
    {
      header.size = (UInt32)Marshal.SizeOf(typeof(PingMessage));
      header.id = 0xdeaddead;
      timestamp = (UInt64)System.Diagnostics.Stopwatch.GetTimestamp();
    }
  }

  [StructLayout(LayoutKind.Sequential, Pack = 1)]
  public struct PongMessage
  {
    public Header header;
    public UInt64 ping_timestamp;

    public void Init()
    {
      header.size = (UInt32)Marshal.SizeOf(typeof(PongMessage));
      header.id = 0xdeadbeef;
    }
  }
}
  public class Client
  {
    private Windows.Networking.Sockets.StreamSocket m_socket;
    private System.Threading.Tasks.Task m_connectionTask;
    private Windows.Storage.Streams.DataWriter m_writer;
    private System.IO.Stream m_inputStream;

    private Windows.Networking.HostName m_hostname;
    private string m_port;

    private bool m_connected = false;
    private System.Threading.CancellationTokenSource m_cancellationSource = new System.Threading.CancellationTokenSource();

    private byte[] m_header = new byte[Marshal.SizeOf(typeof(Messages.Header))];
    private byte[] m_messageBuffer = null;

    //private BlockingRingBuffer<byte[]> m_receiveQueue = new BlockingRingBuffer<byte[]>(1024 * 1024, true);

    public static byte[] Serialize<T>(ref T str)
    {
      int size = Marshal.SizeOf(str);
      byte[] arr = new byte[size];
      GCHandle h = default(GCHandle);
      try
      {
        h = GCHandle.Alloc(arr, GCHandleType.Pinned);
        Marshal.StructureToPtr<T>(str, h.AddrOfPinnedObject(), false);
      }
      finally
      {
        if (h.IsAllocated)
        {
          h.Free();
        }
      }
      return arr;
    }

    public static T Deserialize<T>(byte[] arr) where T : struct
    {
      T str = default(T);
      GCHandle h = default(GCHandle);
      try
      {
        h = GCHandle.Alloc(arr, GCHandleType.Pinned);
        str = (T)Marshal.PtrToStructure(h.AddrOfPinnedObject(), typeof(T));
      }
      finally
      {
        if (h.IsAllocated)
        {
          h.Free();
        }
      }
      return str;
    }

    public async void Send(byte[] bytes)
    {
      await m_connectionTask;

      if (!m_connected)
      {
        Console.WriteLine("NOT CONNECTED: dropping message");
        return;
      }

      try
      {
        m_writer.WriteBytes(bytes);
        await m_writer.StoreAsync();
        Console.WriteLine("Sent " + bytes.Length + " bytes");
      }
      catch (Exception exception)
      {
        // If this is an unknown status it means that the error if fatal and retry will likely fail.
        if (Windows.Networking.Sockets.SocketError.GetStatus(exception.HResult) == Windows.Networking.Sockets.SocketErrorStatus.Unknown)
        {
          throw;
        }
        Console.WriteLine("Error: Failed to send: " + exception.Message);
      }
    }

    /*
    public byte[] GetNextMessage()
    {
      byte[] buffer;
      m_receiveQueue.TryDequeue(out buffer);
      return buffer;
    }
    */

    private async void ReadMessages()
    {
      //TODO: handle disconnections and exceptions?
      while (true)
      {
        // Read header
        await m_inputStream.ReadAsync(m_header, 0, m_header.Length, m_cancellationSource.Token);
        if (m_cancellationSource.IsCancellationRequested)
        {
          return;
        }

        // Read remainder of message
        Messages.Header header = Deserialize<Messages.Header>(m_header);
        m_messageBuffer = new byte[header.size];
        Buffer.BlockCopy(m_header, 0, m_messageBuffer, 0, m_header.Length);
        await m_inputStream.ReadAsync(m_messageBuffer, m_header.Length, (int)header.size - m_header.Length, m_cancellationSource.Token);
        if (m_cancellationSource.IsCancellationRequested)
        {
          return;
        }

        // Enqueue
        //TODO: put on ring buffer
        Console.WriteLine("Read message: {0:x}, {1} bytes, thread={2}", header.id, header.size, System.Threading.Thread.CurrentThread.ManagedThreadId);
        m_messageBuffer = null;
      }
    }

    public async void Connect(bool blockUntilConnected, Action<bool, Exception> OnConnected = null)
    {
      m_connectionTask = System.Threading.Tasks.Task.Run(
        async () =>
        {
          Exception e = null;

          try
          {
            await m_socket.ConnectAsync(m_hostname, m_port);
            m_writer = new Windows.Storage.Streams.DataWriter(m_socket.OutputStream);
            m_inputStream = m_socket.InputStream.AsStreamForRead();
            m_connected = true;
            Console.WriteLine("Remote Address: " + m_socket.Information.RemoteAddress);
          }
          catch (Exception exception)
          {
            Console.WriteLine("Error: Failed to connect: " + exception.Message);
            m_connected = false;
            e = exception;
          }

          if (OnConnected != null)
          {
            OnConnected(m_connected, e);
          }
        }
      );

      if (blockUntilConnected)
      {
        m_connectionTask.Wait();
      }

      await m_connectionTask;

      if (m_connected)
      {
        ReadMessages();
      }
    }

    public Client(string hostname, int port)
    {
      m_socket = new Windows.Networking.Sockets.StreamSocket();
      m_hostname = new Windows.Networking.HostName(hostname);
      m_port = port.ToString();
    }

    ~Client()
    {
      m_cancellationSource.Cancel();
      //m_receiveQueue.StopWaiting(); // queue cannot be reused after this
      m_writer.Dispose();
      m_socket.Dispose();
    }
  }