c1_unity/Assets/Scripts/Common/Network/TCPService/TCPChannel.Receive.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using Facebook.MiniJSON;
using System.Security.Cryptography;

namespace BF
{
    internal partial class TCPChannel
    {
        //incoming message
        private readonly NetSafeQueue<NetIncomingMessage> releasedIncomingMessage = new NetSafeQueue<NetIncomingMessage>(10);
        private readonly NetSafeQueue<NetIncomingMessage> receiveMessageQueue = new NetSafeQueue<NetIncomingMessage>(10);

        private CircularBuffer receiveBuffer;
        private int receiveBodyLength = 0;        //receive optimize
        private NetAesEncrypt aesEncrypt;
        private NetAesEncrypt aesDecrypt;
        /// <summary>
        /// last receive server message seq
        /// </summary>
        private uint receiveSeq = 0;

        private void InitializeReceive()
        {
            receiveSeq = 0;
            receiveBuffer = new CircularBuffer(ownerConnection.configuration.ReceiveBufferCapacity);
        }

        private void ResetReceiveBuffer()
        {
            receiveBodyLength = 0;
            receiveBuffer.Reset();
        }

        private bool IsCanReceiveData()
        {
            return actualStatus == NetConnectStatus.VerifyConnecting || actualStatus == NetConnectStatus.Connected || actualStatus == NetConnectStatus.Reconnecting || actualStatus == NetConnectStatus.Authing;
        }

        private int GetReceiveBufferAvailableSize()
        {
            int size = 0;
            switch (receiveBuffer.EState)
            {
                case CircularBuffer.State.WriteAhead:
                    size = receiveBuffer.Size - receiveBuffer.WritePosition;
                    break;

                case CircularBuffer.State.ReadAhead:
                    size = receiveBuffer.ReadPosition - receiveBuffer.WritePosition;
                    break;

                case CircularBuffer.State.ReadEqualWrite:
                    size = receiveBuffer.Size - receiveBuffer.WritePosition;
                    break;

                case CircularBuffer.State.WriteEqualRead:
                    CircularBuffer newReceiveBuffer = new CircularBuffer(2 * receiveBuffer.Size);
                    int length = receiveBuffer.HowManyCanRead;
                    if (length > 0)
                    {
                        newReceiveBuffer.Write(receiveBuffer.Read(length));
                    }
                    receiveBuffer.Dispose();
                    receiveBuffer = newReceiveBuffer;
                    size = receiveBuffer.Size - length;
                    break;

                default:
                    throw new NotImplementedException();
            }

            return size;
        }

        private void StartReceiveAsync()
        {
            if (!IsCanReceiveData())
            {
                return;
            }

            int size = GetReceiveBufferAvailableSize();
            if (size <= 0)
            {
                LogError(NetErrorCode.ReceiveBufferError, "Receive buffer not enough. Is full ?");
                return;
            }

            try
            {
                innArgs.SocketFlags = SocketFlags.None;
                innArgs.SetBuffer(receiveBuffer.Array, receiveBuffer.WritePosition, size);
            }
            catch (Exception exception)
            {
                throw new NetException(exception.ToString());
            }

            try
            {
                if (socket.ReceiveAsync(innArgs))
                {
                    return;
                }
                OnReceiveComplete(innArgs);
            }
            catch (SocketException exception)
            {
                HandleSocketError(exception.SocketErrorCode, $"Begin receive socket error : {exception}");
            }
            catch (Exception exception)
            {
                LogError(NetErrorCode.ExceptionError,$"Begin receive error : {exception}");
            }
        }

        private void OnReceiveComplete(SocketAsyncEventArgs args)
        {
            if (socket == null)
            {
                return;
            }

            if (!IsCanReceiveData())
            {
                return;
            }

            if (args.SocketError != SocketError.Success)
            {
                //LogError((int)args.SocketError, "Socket receive complete, Error!");
                HandleSocketError(args.SocketError, "Socket receive complete, Error!");
                return;
            }

            //BytesTransferred == 0 ，The remote end has closed the connection
            if (args.BytesTransferred <= 0)
            {
                CreateServerDisconnectMessage();
                return;
            }

            receiveBuffer.MoveWritePosition(args.BytesTransferred);
            if (!ProcessReceiveBuffer())
            {
                //data error
                LogError(NetErrorCode.DataParseError, "Process receive buffer error.");
                return;
            }

            StartReceiveAsync();
        }

        private bool ProcessReceiveBuffer()
        {
            bool success = true;
            while (true)
            {
                if (receiveBodyLength > 0)
                {
                    if (receiveBuffer.HowManyCanRead < receiveBodyLength)
                    {
                        break;
                    }

                    var data = receiveBuffer.Read(receiveBodyLength);
                    var incomingMsg = ownerConnection.CreateIncomingMessage(NetIncomingMessageType.Data, data);
                    EnqueueReceiveMessage(incomingMsg);
                    receiveBodyLength = 0;
                }
                else
                {
                    receiveBodyLength = messageService.ReadBodyLength(receiveBuffer);
                    if (receiveBodyLength == 0)
                    {
                        break;
                    }
                    else if (receiveBodyLength < 0)
                    {
                        success = false;
                        break;
                    }
                }
            }

            return success;
        }

        private void ReceiveUpdate()
        {
            while (receiveMessageQueue.Count > 0)
            {
                NetIncomingMessage incomingMessage = null;
                bool result = receiveMessageQueue.TryDequeue(out incomingMessage);
                if (!result)
                {
                    NetException.Assert(false, "Receive message queue TryDequeue error.");
                    LogWarning("Receive message queue TryDequeue error.");
                    break;
                }

                // BFLog.Log("ReceiveUpdate Length = " + incomingMessage.Data.Length.ToString());
                //首次回包，特殊处理
                if (actualStatus == NetConnectStatus.VerifyConnecting)
                {
                    HandleFirstReceiveMessage(incomingMessage);
                    return;
                }

                result = messageService.Deserialize(incomingMessage, incomingMessage.Seq);
                if (!result)
                {
                    //deserialize error
                    LogError(NetErrorCode.DataParseError,"Deserialize data error.");
                    break;
                }

                // BFLog.Log($"Receive Message group = {incomingMessage.Group}, cmd = {incomingMessage.CMD}, seq = {incomingMessage.Seq}");

                //心跳包，特殊处理
                if (FilterHeartBeatIncomingMessage(incomingMessage))
                {
                    return;
                }

                //auth消息回包，特殊处理
                if (FilterAuthIncomingMessage(incomingMessage))
                {
                    return;
                }

                //重连消息回包，特殊处理
                if (FilterReconnectIncomingMessage(incomingMessage))
                {
                    return;
                }

                //处理重连成功后，接受丢包的补包消息
                if (FilterMissingDataSyncIncomingMessage(incomingMessage))
                {
                    return;
                }

                if (HandleKickNtfMessage(incomingMessage))
                {
                    return;
                }

                if (HandleOtherDeviceLoginMessage(incomingMessage))
                {
                    return;
                }

                //update last success receive message seq
                //group == 0 不需要接收seq
                if (incomingMessage.Group != 0)
                {
                    receiveSeq = incomingMessage.Seq;
                }

                ReleaseMessage(incomingMessage);

#if BF_DEBUG
                NetStatistics.SetReceiveDataSeq(ownerConnection.configuration.UniqueIdentifier, receiveSeq);
#endif
            }

            UpdateFilterServerDisconnectMessage();
        }

        /// <summary>
        /// 其他人登陆同角色时，老的连接会收到挤号消息并主动断开连接。客户端在收到该消息后应提示玩家并返回登陆界面
        /// </summary>
        /// <param name="incomingMessage"></param>
        /// <returns></returns>
        private bool HandleOtherDeviceLoginMessage(NetIncomingMessage incomingMessage)
        {
            if (incomingMessage.Group != 0 || incomingMessage.CMD != 8)
            {
                return false;
            }

            SetConnectContext(NetConnectContext.Invalid);
            Disconnect();
            LogError(NetErrorCode.OtherDeviceLogin, $"Other Device login..");

            return true;
        }

        /// <summary>
        /// 处理玩家被服务器强制踢下线消息；收到消息后，应立刻断开TCP连接
        /// @TODO logic
        /// </summary>
        /// <param name="incomingMessage"></param>
        private bool HandleKickNtfMessage(NetIncomingMessage incomingMessage)
        {
            if (incomingMessage.Group != 0 || incomingMessage.CMD != 9)
            {
                return false;
            }

            var reason = "KickNtf";
            string str = Encoding.UTF8.GetString(incomingMessage.Data);
            Dictionary<string, object> kickNtfRsp = Json.Deserialize(str) as Dictionary<string, object>;
            if (kickNtfRsp != null)
            {
                reason = (string)kickNtfRsp["reason"];
            }

            SetConnectContext(NetConnectContext.Invalid);
            Disconnect();
            LogError(NetErrorCode.ServerKickNtfClient, $"{reason}");
            return true;
        }

        /// <summary>
        /// 每次Socket连接成功后，接受到第一个包，做如下处理：
        /// 1. 根据收到的包，进行密钥交换；
        /// 2. 如果在重连状态，发送一个重连消息，收到重连成功消息后，进入连接成功状态；否则，在正常连接状态，直接连接成功
        /// </summary>
        /// <param name="incomingMessage"></param>
        private void HandleFirstReceiveMessage(NetIncomingMessage incomingMessage)
        {
            if (connectContext == NetConnectContext.Reconnect)
            {
                SetActualStatus(NetConnectStatus.Reconnecting);
                ExchangeAESKeyWithServer(incomingMessage);
                SendAuthMessage();
            }
            else
            {
                ownerConnection.ChannelConnectVerifySuccess(this);
                SetActualStatus(NetConnectStatus.Authing);
                ExchangeAESKeyWithServer(incomingMessage);
                SendAuthMessage();
            }
        }

        /// <summary>
        /// 首次回包，默认是交换Rc4密钥
        /// </summary>
        /// <param name="incomingMessage"></param>
        private void ExchangeAESKeyWithServer(NetIncomingMessage incomingMessage)
        {
            byte[] aesKey = new byte[incomingMessage.Data.Length];
            byte[] aesKeyA = new byte[incomingMessage.Data.Length];
            byte[] aesKeyB = new byte[incomingMessage.Data.Length];
            byte[] llaveB = System.Text.Encoding.ASCII.GetBytes(rsaKey);
            NetAesEncrypt tmpAesEncrypt = new NetAesEncrypt(llaveB);
            tmpAesEncrypt.aesKey = System.Text.Encoding.ASCII.GetBytes(rsaKey);
            tmpAesEncrypt.Decrypt(incomingMessage.Data, 0, incomingMessage.Data.Length, aesKey, 0);

            Array.Copy(aesKey, aesKeyA, aesKey.Length);
            Array.Copy(aesKey, aesKeyB, aesKey.Length);

            aesEncrypt = new NetAesEncrypt(aesKeyA);
            aesEncrypt.aesKey = aesKey;
            messageService.SetEncrypt(aesEncrypt);

            aesDecrypt = new NetAesEncrypt(aesKeyB);
            aesDecrypt.aesKey = aesKey;
            messageService.SetDecrypt(aesDecrypt);
        }

        private void EnqueueReceiveMessage(NetIncomingMessage incomingMessage)
        {
            NetException.Assert(incomingMessage.MessageType != NetIncomingMessageType.Error);

#if BF_DEBUG
            string connectId = ownerConnection.configuration.UniqueIdentifier;
            var result = NetStatistics.CheckEnableReceiveData(connectId);
            if (result)
            {
                receiveMessageQueue.Enqueue(incomingMessage);
            }
            else
            {
                NetStatistics.EnqueueCacheReceiveMessage(connectId, incomingMessage);
            }
#else
            receiveMessageQueue.Enqueue(incomingMessage);
#endif
        }

        private void ReleaseMessage(NetIncomingMessage message)
        {
            NetException.Assert(message.MessageType != NetIncomingMessageType.Error);

            releasedIncomingMessage.Enqueue(message);
        }

    }

}