server/CSserver/csserver/server/Peer.cs

using ProtoDataBuff;
using System;
using System.IO;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Threading.Channels;
using System.Threading.Tasks;
using Chat;
using pb = global::Google.Protobuf;
namespace ChatServer.server
{
    /// <summary>
    /// 客户端状态
    /// </summary>
    enum ClientState
    {
        /// <summary>
        /// 已连接,未登录
        /// </summary>
        Connected,
        /// <summary>
        /// 已登录
        /// </summary>
        Logined,
        /// <summary>
        /// 已离开
        /// </summary>
        Leaved
    }
    class Peer
    {
        /// <summary>
        /// 客户端状态字段
        /// </summary>
        public ClientState CurrentState = ClientState.Connected;

        /// <summary>
        /// peer的唯一Id
        /// </summary>
        public readonly int Id;

        /// <summary>
        /// socket连接
        /// </summary>
        public readonly Socket Sock;
        /// <summary>
        /// 玩家ID
        /// </summary>
        public string UID { get; private set; } = "";
        /// <summary>
        /// 分区id
        /// </summary>
        public int zoneid { get; private set; } = 0;
        public string Name { get; private set; } = "";
        /// <summary>
        /// 线程安全的peer计数器
        /// </summary>
        private volatile static int _UniqPeerId = 0;
        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="sock"></param>
        public Peer(Socket sock)
        {
            this.Sock = sock;
            Sock.ReceiveTimeout = 1500;                                         // 接收等待超时时间设为1.5秒 
            Id = _UniqPeerId++;
            var t = Task.Run(() => recv(Sock));                                 // 新建接收线程 
            var tcs = Task.Run(WriteToClient);                                  // 向客户端发送消息线程 
        }

        public string Key => UID + ":" + zoneid;
        public void OnLogin(sSocketData data)
        {
            var msg = CS_ChatLogin.Parser.ParseFrom(data._data);
            if (msg != null)
            {
                this.UID = msg.Uid;
                this.zoneid = msg.Zoneid;
                this.Name = msg.Name;
                Program.AddPeer(this);
                this.SendEvent(eProtocalCommand.ScChatLogin, new SC_ChatLogin() { Code = SC_ChatLogin.Types.ErrorCode.Ok });
            }
        }
        public void Close()
        {
            this.CurrentState = ClientState.Leaved;                              // 关闭接收窗口
            Program.RemovePeer(this.Key);                                      // 从房间中移除自己 
            this.sendDataBuffer.Reader.Completion
                .ContinueWith(t =>
                Task.Delay(5000).ContinueWith(t1 => Sock.Close())
                );                                                               // 延迟5秒关闭套接字
        }


        /// <summary>
        /// 向客户端发送事件
        /// </summary>
        public void SendEvent(eProtocalCommand msgType, pb::IMessage msg)
        {
            SendToClient(msgType, msg);
        }

        /// <summary>
        /// 向客户端发送消息
        /// </summary>
        /// <param name="msgType"></param>
        /// <param name="msg"></param>
        void SendToClient(eProtocalCommand msgType, pb::IMessage msg)
        {
            using var ms = new MemoryStream();
            using var os = new pb::CodedOutputStream(ms);
            msg.WriteTo(os);
            os.Flush();
            ms.Seek(0, SeekOrigin.Begin);
            var sdata = sSocketData.FromBytes(msgType, ms.ToArray());
            sendDataBuffer.Writer.WriteAsync(sdata).AsTask().Wait();
        }


        /// <summary>
        /// 向客户端写入消息
        /// </summary>
        async void WriteToClient()
        {
            while (true)
            {
                try
                {
                    var msg = await sendDataBuffer.Reader.ReadAsync();
                    await Sock.SendAsync(new ArraySegment<byte>(msg.ToBytes()), SocketFlags.None);
                }
                catch (Exception)
                {
                    break;
                }

            }
            Close();
        }


        /// <summary>
        /// 发送buffer
        /// </summary>
        private Channel<sSocketData> sendDataBuffer = Channel.CreateUnbounded<sSocketData>();


        /// <summary>
        /// 接收客户端发来的信息，客户端套接字对象
        /// </summary>
        /// <param name="socketclientpara"></param>    
        async void recv(Socket socketServer)
        {

            var _databuffer = new DataBuffer();
            byte[] arrServerRecMsg = new byte[4096];                                       // 创建一个内存缓冲区，其大小为4k字节  
            while (true)
            {
                try
                {
                    var length = await socketServer.ReceiveAsync(new ArraySegment<byte>(arrServerRecMsg), SocketFlags.None);   // 将接收到的信息存入到内存缓冲区，并返回其字节数组的长度    
                    if (length <= 0 || CurrentState == ClientState.Leaved)                                                     // 视为客户端已经close连接
                    {
                        break;
                    }
                    _databuffer.AddBuffer(arrServerRecMsg, length);                                  //将收到的数据添加到缓存器中
                    while (_databuffer.GetData(out sSocketData _socketData))                         //取出一条完整数据
                    {
                        if (eProtocalCommand.CsChatLogin == _socketData._protocallType)
                        {
                            OnLogin(_socketData);
                            continue;
                        }
                        await Program.MsgChannel.Writer.WriteAsync(new KeyValuePair<string, sSocketData>(this.Key, _socketData));    // 放入channel
                    }
                    Array.Clear(arrServerRecMsg, 0, length);
                }
                catch (SocketException e)
                {
                    if (e.ErrorCode == 10060)                                          // 超时的时候错误号码是10060
                    {
                        continue;                                                      // 继续等待  
                    }
                    break;
                }
                catch (Exception)
                {
                    break;
                }
            }

            Close();                                                                  // 关闭
        }

    }
}