hjha-server/ServerCore/NetWork/TChannel.cs

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();

        /// <summary>
        /// 接受数据的缓存
        /// </summary>
        private readonly CircularBuffer recvBuffe = new CircularBuffer();

        /// <summary>
        /// 发送数据的缓存
        /// </summary>
        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}");
        }
    }
}