rg1/server/Sog/Core/Net/NetSession.cs

/*
 Sog 游戏基础库
 2016 by zouwei
 */


using System;
using System.Collections.Concurrent;
using System.Collections.Generic;

using System.Net;
using System.Net.Sockets;
using System.Threading;
using Sog.IO;


namespace Sog
{
    /// <summary>
    /// NetSession
    /// </summary>
    public abstract class NetSession
    {
        const int BUFFER_LENGTH_MAX = 1024 * 1024 * 100;


        public Socket WorkSocket => m_socket;
        public EndPoint RemoteEndPoint => m_remoteEndPoint;

        public long SessionID { get; private set; }
        public bool IsSocketClosed { get; private set; }


        protected object m_socketLocker = new object();
        protected Socket m_socket;
        protected EndPoint m_remoteEndPoint;


        internal long LastRecvDataTimeSecond;
        internal long LastSendDataTimeSecond;
        protected int m_noWriteableCheckCount;


        protected Queue<MessageData> m_recvMessageQueue = new Queue<MessageData>();
        protected object m_recvMessageQueueLocker = new object();
        protected Queue<MessageData> m_recvBigDataMessageQueue = new Queue<MessageData>();

        protected Queue<MessageData> m_sendMessageQueue = new Queue<MessageData>();
        protected int m_dataLengthInSendQueue;

        // 异步模式先发送到Async队列
        protected Queue<MessageData> m_sendMessageAsyncQueue = new Queue<MessageData>();
        protected object m_sendMessageAsyncQueueLocker = new object();


//        protected bool m_recvAsyncPending;
//        protected bool m_sendAsyncPending;
//        protected object m_recvAsyncLocker;
//        protected object m_sendAsyncLocker;
//        protected SocketAsyncEventArgs m_recvAsyncEventArgs;
//        protected SocketAsyncEventArgs m_sendAsyncEventArgs;


        protected byte[] m_sendBuffer;
        protected int m_sendDataLeft;
        protected int m_sendDataPos;
        

        protected byte[] m_recvBuffer;
        protected int m_alreadyRecvLength;


        //长度，消息头信息
        protected byte[] m_headerBytes;
        protected byte[] m_zeroLengthBytes;

        //消息头的长度
        protected int m_minHeaderLength;
        protected ProtoCSStructPacker m_protoPacker;

        public bool WriteSendRecvLog { get; set; }

        public long TotalRecvLength;
        public long TotalSendLength;


        public abstract void WriteWriteableSocket();

        public abstract void RecvReadableSocket();

        public abstract void StartSendAsync();

        public abstract void StartRecvAsync();

        protected abstract void OnBindSocket();

        protected abstract void CloseSocket();

        public abstract void CheckSocketAndSendKeepAlive();

        /// <summary>
        /// 异步模式下接收到数据事件
        /// </summary>
        public event NetEventHandler AsyncModeDataReceived;
        private void OnAsyncModeDataReceived(SessionEventArgs e)
        {
            TraceLog.TraceDetail("NetSession.OnAsyncModeDataReceived remote {0} session {1} length {2}"
                , e.Session.RemoteEndPoint, e.Session.SessionID, e.Message.Header.Length);

            if (AsyncModeDataReceived != null)
            {
                AsyncModeDataReceived(null, e);
            }
        }

        /// <summary>
        /// </summary>
        /// <param name="socket">Socket.</param>
        public NetSession(Socket socket)
        {
            m_protoPacker = ProtoPackerFactory.Instance.GetProtoCSStructPacker();
            m_minHeaderLength = m_protoPacker.GetHeaderLength();

            m_headerBytes = new byte[m_protoPacker.GetFullHeaderLength()];
            m_zeroLengthBytes = new byte[8];

            SessionID = SessionIDUtil.GenID();
            m_socket = socket;
        }


        /// <summary>
        /// 重新设置sessionid，gateserver会用到
        /// </summary>
        /// <param name="id"></param>
        public void ResetSessionID(long id)
        {
            SessionID = id;
        }

        /// <summary>
        /// 设置读写缓冲大小，系统缺省是65k
        /// </summary>
        /// <param name="iNewBufferSize"></param>
        public void SetSocketBufferSize(int iNewBufferSize)
        {
            //设置bufersize
            int iOldBufferSize = m_socket.SendBufferSize;

            m_socket.SendBufferSize = iNewBufferSize;
            m_socket.ReceiveBufferSize = iNewBufferSize;

            TraceLog.Debug("NetSession.SetSocketBufferSize socket set new buffersize {0}, old size {1}", iNewBufferSize, iOldBufferSize);
        }

        // bind只会发生在连接成功建立后
        public void Bind(Socket socket)
        {
            if (socket == null)
            {
                TraceLog.Error("NetSession.Bind session {0} socket is null", SessionID);
                return;
            }

            if (m_socket != null && m_socket != socket)
            {
                TraceLog.Error("NetSession.Bind session {0} already bind socket, close first", SessionID);
                Close();    
            }

            lock (m_socketLocker)
            {
                m_socket = socket;
                IsSocketClosed = false;

                OnBindSocket();
               
                // 如果有正在发送的数据，直接丢弃
                m_sendDataLeft = 0;
                m_sendDataPos = 0;
            }
        }

        public void Close()
        {
            TraceLog.Trace("NetSession.Close session {0} remote {1}", SessionID, RemoteEndPoint);

            if (IsSocketClosed)
            {
                return;
            }

            lock (m_socketLocker)
            {
                if (IsSocketClosed)
                {
                    return;
                }

                CloseSocket();
            }

            // todo 关闭前要不要尝试再解析一次
//            TryReadMessageFromBuffer(false, false);

            // 如果有正在接收的数据，直接丢弃
            m_alreadyRecvLength = 0;

            //如果有正在发送的数据，直接丢弃
            m_sendDataLeft = 0;
            m_sendDataPos = 0;

            IsSocketClosed = true;  
        }

        protected void TryReadMessageFromBuffer(bool threadSafe, bool raiseAsyncModeEvent)
        {
            if(m_alreadyRecvLength < m_minHeaderLength)
            {
                return;
            }

            int iBufferLeftLength = m_alreadyRecvLength;
            int iBufferStartPos = 0;

            while (iBufferLeftLength >= m_minHeaderLength)
            {
                MessageHeader header = new MessageHeader();
                int headLen = m_protoPacker.UnPackHeader(m_recvBuffer, iBufferStartPos, iBufferLeftLength, ref header);

                //没有完整消息头
                if (headLen == 0)
                {
                    TraceLog.Trace("NetSession.TryReadMessageFromBuffer no enougth head, BufferLeftLength {0} BufferStartPos {1}"
                        , iBufferLeftLength, iBufferStartPos);
                    break;
                }

                int dataLen = header.Length;

                //判断是否合法
                if(dataLen < 0 || dataLen > 64*1024 - 24)
                {
                    //主动断线
                    TraceLog.Error("NetSession.TryReadMessageFromBuffer recv a invalid length {0} packet, close session {1}, remote {2}",
                        dataLen, SessionID, RemoteEndPoint);

                    Close();
                    return;
                }
                

                //有完整的消息了
                if (dataLen + headLen <= iBufferLeftLength)
                {
                    //注意消息长度可以为0，就是只有消息头的消息，没有实际内容
                    //类型为0，长度为0的消息不处理
                    if (dataLen >= 0 && header.Type > 0)
                    {
                        MessageData message = new MessageData();
                        message.MallocData(dataLen);
                        Buffer.BlockCopy(m_recvBuffer, iBufferStartPos + headLen, message.Buffer.Data, 0, dataLen);
                       
                        message.Header = header;

                        if(header.Type == (int)SpecialMessageType.BigMessageStart
                            || header.Type == (int)SpecialMessageType.BigMessageTrans)
                        {
                            m_recvBigDataMessageQueue.Enqueue(message);

                            if(threadSafe)
                            {
                                lock(m_recvMessageQueueLocker)
                                {
                                    BigDataMessage.TryGetFullBigDataMessage(m_recvBigDataMessageQueue, m_recvMessageQueue);
                                }
                            }
                            else
                            {
                                BigDataMessage.TryGetFullBigDataMessage(m_recvBigDataMessageQueue, m_recvMessageQueue);
                            }
                            
                        }
                        else
                        {
                            if(threadSafe)
                            {
                                lock(m_recvMessageQueueLocker)
                                {
                                    m_recvMessageQueue.Enqueue(message);
                                }
                            }
                            else
                            {
                                m_recvMessageQueue.Enqueue(message);
                            }
                            
                            if (WriteSendRecvLog)
                            {
                                TraceLog.TraceDetail("NetSession.TryReadMessageFromBuffer remote {0} session {1} recv type {2} length {3}"
                                    , RemoteEndPoint, SessionID, header.Type, header.Length);

                                //TraceLog.Trace("data content {0}", StringUtils.ToHexString(data));
                            }
                        }
                    }

                    iBufferStartPos += dataLen + headLen;
                    iBufferLeftLength -= dataLen + headLen;
                }
                else
                {
                    TraceLog.TraceDetail("NetSession.TryReadMessageFromBuffer no full message, BufferLeftLength {0} BufferStartPos {1}"
                        , iBufferLeftLength, iBufferStartPos);
                    break;
                }
            }

            if(iBufferLeftLength > 0 && iBufferStartPos > 0)
            {
                //剩下的内容向前移动
                Buffer.BlockCopy(m_recvBuffer, iBufferStartPos, m_recvBuffer, 0, iBufferLeftLength);
            }

            m_alreadyRecvLength = iBufferLeftLength;

            if(raiseAsyncModeEvent && m_recvMessageQueue.Count > 0)
            {
                while(m_recvMessageQueue.Count > 0)
                {
                    MessageData message = m_recvMessageQueue.Dequeue();

                    SessionEventArgs e = new SessionEventArgs();
                    e.Message = message;
                    e.Session = this;
  
                    OnAsyncModeDataReceived(e);
                }
            }
        }

        public Queue<MessageData> GetReceivedDataQueue()
        {
            return m_recvMessageQueue;
        }

        protected void CopySendBuffer()
        {
            //只有buffer的数据全部发送完毕后才行
            if (m_sendDataLeft > 0)
            {
                return;
            }

            if(m_sendMessageAsyncQueue.Count > 0)
            {
                lock(m_sendMessageAsyncQueueLocker)
                {
                    while(m_sendMessageAsyncQueue.Count > 0)
                    {
                        MessageData message = m_sendMessageAsyncQueue.Dequeue();
                        //调用这个，大消息需要拆分下
                        //线程安全，也不会有顺序问题
                        SendMessageToQueue(message);
                    }
                }
            }

            while (m_sendMessageQueue.Count > 0)
            {
                MessageData message = m_sendMessageQueue.Dequeue();
                m_dataLengthInSendQueue -= message.Header.Length;

                int headerLength = m_protoPacker.PackHeader(message.Header, m_headerBytes);
                
                Buffer.BlockCopy(m_headerBytes, 0, m_sendBuffer, m_sendDataLeft, headerLength);
                m_sendDataLeft += headerLength;

                Buffer.BlockCopy(message.Buffer.Data, 0, m_sendBuffer, m_sendDataLeft, message.Buffer.Length);
                m_sendDataLeft += message.Buffer.Length;

                message.FreeData();

                //查看一下，下个消息长度能否copy成功,不行就跳出
                if (m_sendMessageQueue.Count > 0 )
                {
                    message = m_sendMessageQueue.Peek();
                    if(m_sendDataLeft + message.Buffer.Length + 100 >= m_sendBuffer.Length)
                    {
                        break;
                    }
                }

            }
        }

        public bool IsFull()
        {
            return m_dataLengthInSendQueue >= BUFFER_LENGTH_MAX;
        }

        /// <summary>
        /// 是否有数据需要发送
        /// </summary>
        /// <returns></returns>
        public bool IsHaveDataNeedSend()
        {
            if (m_socket == null)
            {
                return false;
            }

            if (m_sendMessageQueue.Count > 0 || m_sendMessageAsyncQueue.Count > 0)
            {
                return true;
            }

            return m_sendDataLeft > 0;
        }

        public void SendMessageToAsyncQueue(MessageData message)
        {
            lock(m_sendMessageAsyncQueueLocker)
            {
                m_sendMessageAsyncQueue.Enqueue(message);
            }
        }

        public void SendMessageToQueue(MessageData message)
        {
            if (IsFull())
            {
                TraceLog.Error("send buff full drop msg type {0}, length {1} remote {2}",
                    message.Header.Type,
                    message.Header.Length,
                    m_remoteEndPoint.ToString());

                return;
            }

            //大消息需要拆开
            if(message.Header.Length > BigDataMessage.MessageLenghMinNeedSplit)
            {
                List<MessageData> splits = BigDataMessage.SplitMessage(message);
                foreach(var one in splits)
                {
                    m_sendMessageQueue.Enqueue(one);
                    m_dataLengthInSendQueue += one.Header.Length;
                }
                message.FreeData();
            }
            else
            {
                m_sendMessageQueue.Enqueue(message);
                m_dataLengthInSendQueue += message.Header.Length;
            }
        }

        /// <summary>
        /// 逻辑层不要调用，底层用
        /// </summary>
        /// <param name="message"></param>
        internal void DirectSend(MessageData message)
        {
            m_sendDataLeft = 0;
            m_sendDataPos = 0;

            int headerLength = m_protoPacker.PackHeader(message.Header, m_headerBytes);

            Buffer.BlockCopy(m_headerBytes, 0, m_sendBuffer, m_sendDataLeft, headerLength);
            m_sendDataLeft += headerLength;
            if (message.Buffer.Data != null)
            {
                Buffer.BlockCopy(message.Buffer.Data, 0, m_sendBuffer, m_sendDataLeft, message.Buffer.Length);
                m_sendDataLeft += message.Buffer.Length;
            }
        }

        public void StartSendAsyncMessage(MessageData message)
        {
            SendMessageToAsyncQueue(message);
            StartSendAsync();
        }
    }
}