ClientSocket.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;
using GameServer.net.messaging;
using GameServer.net.messaging.incoming;
using GameServer.net.messaging.outgoing;
using ROTF.Model;
using ROTF.Model.IO;

namespace GameServer.net
{
    public class ClientSocket
    {
        public Socket Socket { get; }
        public Client Client { get; }

        private readonly ConcurrentQueue<byte[]> pendingMessages;

        private bool sending;
        private bool disconnecting;
        private readonly object sendLock = new object();

        public ClientSocket(Socket skt, Client client)
        {
            pendingMessages = new ConcurrentQueue<byte[]>();
            Socket = skt;
            Client = client;
        }

        public void Start()
        {
            receiveMessageLength();
        }

        private void receiveMessageLength()
        {
            var buffer = new byte[4];
            disposeSafeSocketOperation(() => Socket.BeginReceive(buffer, 0, 4, SocketFlags.None, receiveMessageId, buffer));
        }

        private void receiveMessageId(IAsyncResult ar)
        {
            disposeSafeSocketOperation(() => Socket.EndReceive(ar));
            var buffer = (byte[])ar.AsyncState;
            Array.Resize(ref buffer,
                IPAddress.NetworkToHostOrder(BitConverter.ToInt32(buffer, 0)));
            disposeSafeSocketOperation(() => Socket.BeginReceive(buffer, 4, 1, SocketFlags.None, receiveMessage, buffer));
        }

        private void receiveMessage(IAsyncResult ar)
        {
            disposeSafeSocketOperation(() => Socket.EndReceive(ar));
            var buffer = (byte[])ar.AsyncState;
            disposeSafeSocketOperation(() => Socket.BeginReceive(buffer, 5, buffer.Length - 5, SocketFlags.None, endMessageReceive, buffer));
        }

        private void endMessageReceive(IAsyncResult ar)
        {
            disposeSafeSocketOperation(() => Socket.EndReceive(ar));
            var buffer = (byte[])ar.AsyncState;

            var b = new byte[buffer.Length-5];

            Buffer.BlockCopy(buffer, 5, b, 0, b.Length);

            var msg = Singleton<MessageMap>.Instance.Require(buffer[4]) as IncomingMessage;
            msg.ReadFromInput(new DataReader(new MemoryStream(b)));

            var handler = Singleton<MessageMap>.Instance.GetHandler(buffer[4]);
            handler.Init(Client);
            handler.Handle(msg);

            receiveMessageLength();
        }

        public void SendMessage(OutgoingMessage message)
        {
            if (disconnecting) return;

            //Structure:
            //
            // Message Length: Int32 (4 bytes) [5 + messageData bytes count]
            // Message Id:     Byte (1 byte)
            // Message Data:   Bytes (x bytes) [RC4 Encrypted]

            Task.Factory.StartNew(() =>
            {
                var msgStrm = new MemoryStream();
                var msgWtr = new DataWriter(msgStrm);
                message.WriteToOutput(msgWtr);

                var messageDataBytes = msgStrm.ToArray();
                var len = 5 + messageDataBytes.Length;
                var messageBytes = new byte[len];

                unsafe
                {
                    var bytes = new byte[4];
                    fixed (byte* b = bytes)
                        *(int*)b = IPAddress.HostToNetworkOrder(len);

                    messageBytes[0] = bytes[0];
                    messageBytes[1] = bytes[1];
                    messageBytes[2] = bytes[2];
                    messageBytes[3] = bytes[3];
                }

                messageBytes[4] = message.MessageId;

                Buffer.BlockCopy(messageDataBytes, 0, messageBytes, 5, len - 5);
                pendingMessages.Enqueue(messageBytes);

                sendPendingMessages();
            });
        }

        private void sendPendingMessages()
        {
            if (pendingMessages.Count == 0)
            {
                if (disconnecting)
                    Disconnect(disconnecting);
                return;
            }

            lock (sendLock)
            {
                if (sending)
                    return;
                sending = true;
            }

            byte[] data;
            var res = pendingMessages.TryDequeue(out data);

            if (!res) return;

            disposeSafeSocketOperation(() => Socket.BeginSend(data, 0, data.Length, SocketFlags.None, sendComplete, data.Length));
        }

        private void sendComplete(IAsyncResult ar)
        {
            SocketError result;
            var bytesTransfered = Socket.EndSend(ar, out result);

            lock (sendLock)
                sending = false;

            if (bytesTransfered != (int)ar.AsyncState)
                throw new InvalidDataException($"Not all bytes were sent!\nBytes: {bytesTransfered}\nSent: {ar.AsyncState}");
            sendPendingMessages();
        }

        public void Disconnect(bool immediate)
        {
            if (immediate)
            {
                Socket.Close();
                Socket.Dispose();
                return;
            }

            disconnecting = true;
        }


        private void disposeSafeSocketOperation(Action action)
        {
            try
            {
                if (Socket.Connected)
                    action();
            }
            catch (ObjectDisposedException)
            {
                Socket.Disconnect(false);
                Socket.Dispose();
            }
        }
    }
}