using System; using System.Diagnostics; using System.IO; using System.Net; using System.Net.Sockets; using System.Threading; using NetWorkMessage; using Server.Core; using Debug = MrWu.Debug.Debug; namespace Server.Net { public sealed class TChannel : AChannel { #if DEBUG public static int TChannelCnt; #endif public int Error { get; set; } private readonly TService Service; private Socket socket; private SocketAsyncEventArgs innArgs = new SocketAsyncEventArgs(); private SocketAsyncEventArgs outArgs = new SocketAsyncEventArgs(); /// /// 接受数据的缓存 /// private readonly CircularBuffer recvBuffe = new CircularBuffer(); /// /// 发送数据的缓存 /// private readonly CircularBuffer sendBuffer = new CircularBuffer(); private bool isSending; private bool isConnected; private readonly PacketParser parser; private readonly byte[] sendCache = new byte[18]; private void OnComplete(object sender, SocketAsyncEventArgs e) { this.Service.Queue.Enqueue(new TArgs() { ChannelId = this.Id, SocketAsyncEventArgs = e }); } public TChannel(long id, IPEndPoint ipEndPoint, TService service) { this.Service = service; this.ChannelType = ChannelType.Connect; this.Id = id; this.socket = new Socket(ipEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp); this.socket.NoDelay = true; this.parser = new PacketParser(this.recvBuffe, this.Service); this.innArgs.Completed += this.OnComplete; this.outArgs.Completed += this.OnComplete; this.RemoteAddress = ipEndPoint; this.isConnected = false; this.isSending = false; this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.Connect, ChannelId = this.Id }); Debug.Info($"创建TChannel:{id} {RemoteAddress}"); #if DEBUG Interlocked.Increment(ref TChannelCnt); #endif } public TChannel(long id, Socket socket, TService service) { this.Service = service; this.ChannelType = ChannelType.Accept; this.Id = id; this.socket = socket; this.socket.NoDelay = true; this.parser = new PacketParser(this.recvBuffe, this.Service); this.innArgs.Completed += this.OnComplete; this.outArgs.Completed += this.OnComplete; this.RemoteAddress = (IPEndPoint)socket.RemoteEndPoint; this.isConnected = true; this.isSending = false; this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.StartRecv, ChannelId = this.Id }); this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.StartSend, ChannelId = this.Id }); Debug.Info($"创建TChannel:{id} {RemoteAddress}"); #if DEBUG Interlocked.Increment(ref TChannelCnt); #endif } #if DEBUG ~TChannel() { Interlocked.Decrement(ref TChannelCnt); } #endif public override void Dispose() { if (this.IsDisposed) { return; } Debug.Info($"channel dispose:{this.Id} {this.RemoteAddress} {this.Error}"); long id = this.Id; this.Id = 0; this.Service.Remove(id); this.socket.Close(); this.innArgs.Dispose(); this.outArgs.Dispose(); this.innArgs = null; this.outArgs = null; this.socket = null; } public void Send(MemoryBuffer stream) { if (this.IsDisposed) { Debug.Log("TChannel Is Disposed!"); throw new Exception("TChannel已经被Dispose,不能发送消息"); } switch (this.Service.ServiceType) { case ServiceType.Actor: { int messageSize = (int)(stream.Length - stream.Position); if (messageSize > ushort.MaxValue * 16) { throw new Exception($"send packet too large:{stream.Length} {stream.Position}"); } //这里是处理包头 //包头 this.sendCache.WriteTo(0, messageSize); //写入包大小 this.sendBuffer.Write(this.sendCache, 0, PacketParser.ActorPacketSizeLength); //Debug.Log($"写入包大小:{messageSize}"); } break; case ServiceType.Router: { int messageSize = (int)(stream.Length - stream.Position); this.sendCache.WriteTo(0, messageSize); this.sendBuffer.Write(this.sendCache, 0, PacketParser.RouterPacketSizeLength); //Debug.Log($"写入包大小:{messageSize} {stream.Position} {stream.Length - stream.Position}"); } break; } this.sendBuffer.Write(stream.GetBuffer(), (int)stream.Position, (int)(stream.Length - stream.Position)); if (!this.isSending) { //Debug.Log("启动发送!"); this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.StartSend, ChannelId = this.Id }); } this.Service.Recycle(stream); } public void ConnectAsync() { Debug.Log($"开始链接:{this.RemoteAddress}"); this.outArgs.RemoteEndPoint = this.RemoteAddress; if (this.socket.ConnectAsync(this.outArgs)) { return; } //同步连接完成 OnConnectComplete(this.outArgs); } public void OnConnectComplete(SocketAsyncEventArgs e) { if (this.socket == null) { return; } if (e.SocketError != SocketError.Success) { this.OnError((int)e.SocketError); return; } Debug.Log("链接完成!"); e.RemoteEndPoint = null; this.isConnected = true; this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.StartRecv, ChannelId = this.Id }); this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.StartSend, ChannelId = this.Id }); } public void OnDisconnectComplete(SocketAsyncEventArgs e) { this.OnError((int)e.SocketError); } public void StartRecv() { while (true) { try { if (this.socket == null) { return; } //最大接收完最后一个数据块 int size = this.recvBuffe.ChunkSize - this.recvBuffe.LastIndex; this.innArgs.SetBuffer(this.recvBuffe.Last, this.recvBuffe.LastIndex, size); //Debug.Log("开始接收!"); } catch (Exception e) { Debug.Error($"TChannel error: {this.Id} {e.Message}"); this.OnError(NetErrorCode.ERR_TChannelRecvError); return; } if (this.socket.ReceiveAsync(this.innArgs)) { //异步接收完成 return; } //同步接收 HandleRecv(this.innArgs); } } public void OnRecvComplete(SocketAsyncEventArgs e) { this.HandleRecv(e); if (this.socket == null) { return; } this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.StartRecv, ChannelId = this.Id }); } private void HandleRecv(SocketAsyncEventArgs e) { if (this.socket == null) { return; } if (e.SocketError != SocketError.Success) { this.OnError((int)e.SocketError); return; } if (e.BytesTransferred == 0) { this.OnError(NetErrorCode.ERR_PeerDisconnect); return; } //Debug.Log("接收完成!"); this.recvBuffe.LastIndex += e.BytesTransferred; //整个块接收完成,创建下一个块 if (this.recvBuffe.LastIndex == this.recvBuffe.ChunkSize) { this.recvBuffe.AddLast(); this.recvBuffe.LastIndex = 0; } while (true) { if (this.socket == null) { return; } try { if (this.recvBuffe.Length == 0) { break; } bool ret = this.parser.Parse(out MemoryBuffer memoryBuffer); if (!ret) //没有一个完整的包 { break; } this.OnRead(memoryBuffer); } catch (Exception exception) { Debug.Error($"ip: {this.RemoteAddress} {exception}"); this.OnError(NetErrorCode.ERR_SocketError); return; } } } public void StartSend() { if (!this.isConnected) { return; } if (this.isSending) { return; } while (true) { try { if (this.socket == null) { this.isSending = false; return; } if (this.sendBuffer.Length == 0) { this.isSending = false; return; } this.isSending = true; int sendSize = this.sendBuffer.ChunkSize - this.sendBuffer.FirstIndex; if (sendSize > this.sendBuffer.Length) { sendSize = (int)this.sendBuffer.Length; } this.outArgs.SetBuffer(this.sendBuffer.First, this.sendBuffer.FirstIndex, sendSize); if (this.socket.SendAsync(this.outArgs)) { //异步发送完成 return; } //同步发送完成 HandleSend(this.outArgs); } catch (Exception e) { throw new Exception( $"socket set buffer error:{this.sendBuffer.First.Length},{this.sendBuffer.FirstIndex}", e); } } } public void OnSendComplete(SocketAsyncEventArgs e) { HandleSend(e); this.isSending = false; this.Service.Queue.Enqueue(new TArgs() { Op = TcpOp.StartSend, ChannelId = this.Id }); } private void HandleSend(SocketAsyncEventArgs e) { if (this.socket == null) { return; } if (e.SocketError != SocketError.Success) { this.OnError((int)e.SocketError); return; } if (e.BytesTransferred == 0) { this.OnError(NetErrorCode.ERR_PeerDisconnect); return; } this.sendBuffer.FirstIndex += e.BytesTransferred; if (this.sendBuffer.FirstIndex == this.sendBuffer.ChunkSize) { this.sendBuffer.FirstIndex = 0; this.sendBuffer.RemoveFirst(); } } private void OnRead(MemoryBuffer memoryBuffer) { try { this.Service.ReadCallBack(this.Id, memoryBuffer); } catch (Exception e) { Debug.Error($"TChannel ReadCallBack: {e}"); this.OnError(NetErrorCode.ERR_PacketParserError); } } private void OnError(int error) { long channelId = this.Id; this.Service.Remove(channelId); this.Service.ErrorCallBack(channelId, error); } public void Log() { Debug.Log($"TChannel:{this.Id} {recvBuffe.Length} {sendBuffer.Length}"); } } }