/* Sog 游戏基础库 2016 by zouwei */ using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Net; using System.Net.Sockets; using System.Threading; namespace SogClient { /// /// NetSession /// public abstract class NetSession { const int BUFFER_LENGTH_MAX = 1024 * 1024 * 100; public Socket WorkSocket => m_socket; public EndPoint RemoteEndPoint => m_remoteEndPoint; public EndPoint LocalEndPoint => m_localEndPoint; public long SessionID { get; private set; } public bool IsSocketClosed { get; private set; } // /// // /// HashCode // /// // public string SessionID { get; private set; } protected object m_socketLocker = new object(); protected Socket m_socket; protected EndPoint m_remoteEndPoint; protected EndPoint m_localEndPoint; internal DateTime LastDataTime; // internal long LastRecvDataTimeSecond; // internal long LastSendDataTimeSecond; protected int m_noWriteableCheckCount; protected Queue m_recvMessageQueue = new Queue(); protected object m_recvMessageQueueLocker = new object(); protected Queue m_recvBigDataMessageQueue = new Queue(); protected Queue m_sendMessageQueue = new Queue(); protected int m_dataLengthInSendQueue; // 异步模式先发送到Async队列 protected Queue m_sendMessageAsyncQueue = new Queue(); protected object m_sendMessageAsyncQueueLocker = new object(); 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 IProtoPacker 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(); /// /// 异步模式下接收到数据事件 /// public event NetEventHandler AsyncModeDataReceived; private void OnAsyncModeDataReceived(SessionEventArgs e) { TraceLog.Trace("NetSession OnAsyncModeDataReceived remote {0} SessionID {1} length {2}" , e.Session.RemoteEndPoint, e.Session.SessionID, e.Message.Header.Length); if (AsyncModeDataReceived != null) { AsyncModeDataReceived(null, e); } } /// /// /// Socket. public NetSession(Socket socket) { m_protoPacker = ProtoPackerFactory.Instance.GetProtoPacker(); m_minHeaderLength = m_protoPacker.GetHeaderLength(); m_headerBytes = new byte[m_protoPacker.GetHeaderLength()]; m_zeroLengthBytes = new byte[8]; m_socket = socket; m_remoteEndPoint = m_socket.RemoteEndPoint; m_localEndPoint = m_socket.LocalEndPoint; } /// /// 重新设置sessionid,gateserver会用到 /// /// public void ResetSessionID(long id) { SessionID = id; } /// /// 设置读写缓冲大小,系统缺省是65k /// /// public void SetSocketBufferSize(int iNewBufferSize) { //设置bufersize int iOldBufferSize = m_socket.SendBufferSize; m_socket.SendBufferSize = iNewBufferSize; m_socket.ReceiveBufferSize = iNewBufferSize; TraceLog.Trace("ClusterChannel.socket_OnConnected 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() { 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; header.Type = 0; header.Length = 0; header.ObjectID = 0; header.ServerID = 0; int headLen = m_protoPacker.UnPackHeader(m_recvBuffer, iBufferStartPos, iBufferLeftLength, ref header); //没有完整消息头 if (headLen == 0) { TraceLog.Trace("TryReadMessageFromBuffer no enougth head ,iBufferLeftLength {0} iBufferStartPos {1} ", iBufferLeftLength, iBufferStartPos); break; } int dataLen = header.Length; //判断是否合法 if(dataLen < 0 || dataLen > 64*1024 - 24) { //主动断线 TraceLog.Error("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) { byte[] data = new byte[dataLen]; Buffer.BlockCopy(m_recvBuffer, iBufferStartPos + headLen, data, 0, dataLen); MessageData message; message.Data = data; 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.Trace("TryReadMessageFromBuffer remote {0} SessionID {1} recv type {2} length {3}" , RemoteEndPoint, SessionID, header.Type, header.Length); TraceLog.Trace("TryReadMessageFromBuffer remote data content {0}", Convert.ToString(message.Data)); } } } iBufferStartPos += dataLen + headLen; iBufferLeftLength -= dataLen + headLen; } else { TraceLog.Trace("TryReadMessageFromBuffer no full message ,iBufferLeftLength {0} iBufferStartPos {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 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(); m_sendMessageQueue.Enqueue(message); m_dataLengthInSendQueue += message.Header.Length; } } } 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.Data, 0, m_sendBuffer, m_sendDataLeft, message.Data.Length); m_sendDataLeft += message.Data.Length; //查看一下,下个消息长度能否copy成功,不行就跳出 if(m_sendMessageQueue.Count > 0 ) { message = m_sendMessageQueue.Peek(); if(m_sendDataLeft + message.Data.Length + 100 >= m_sendBuffer.Length) { break; } } } } public bool IsFull() { return m_dataLengthInSendQueue >= BUFFER_LENGTH_MAX; } /// /// 是否有数据需要发送 /// /// 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 splits = BigDataMessage.SplitMessage(message); foreach(var one in splits) { m_sendMessageQueue.Enqueue(one); m_dataLengthInSendQueue += one.Header.Length; } } else { m_sendMessageQueue.Enqueue(message); m_dataLengthInSendQueue += message.Header.Length; } } /// /// 逻辑层不要调用,底层用 /// /// 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.Data != null) { Buffer.BlockCopy(message.Data, 0, m_sendBuffer, m_sendDataLeft, message.Data.Length); m_sendDataLeft += message.Data.Length; } } public void StartSendAsyncMessage(MessageData message) { SendMessageToAsyncQueue(message); StartSendAsync(); } } }