server/CSserver/BossServer/server/Peer.cs

using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Threading.Channels;
using System.Threading.Tasks;
using ProtoDataBuff;
using pb = global::Google.Protobuf;
namespace BossServer.server
{
    /// <summary>
    /// 客户端状态
    /// </summary>
    enum ClientState
    {
        /// <summary>
        /// 已连接,未登录
        /// </summary>
        Connected,
        /// <summary>
        /// 已登录
        /// </summary>
        Authenticated,
        /// <summary>
        /// 已离开
        /// </summary>
        Leaved
    }


    /// <summary>
    /// 一个连接对象的抽象,代表一个远端的对象
    /// </summary>
    class Peer
    {
        /// <summary>
        /// 玩家属性字段
        /// </summary>
        public Dictionary<string, object> Properties = new();

        /// <summary>
        /// 客户端状态字段
        /// </summary>
        public ClientState CurrentState = ClientState.Connected;


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

        /// <summary>
        /// socket连接
        /// </summary>
        public readonly Socket Sock;

        public string UID
        {
            get
            {
                if (Properties.TryGetValue(PropertyName.Uid, out var uid))
                {
                    return uid.ToString();
                }
                return "";
            }
        }
        public int zoneid
        {
            get
            {
                if (Properties.TryGetValue(PropertyName.Zoneid, out var zoneid))
                {
                    return Convert.ToInt32(zoneid);
                }
                return 1;
            }

        }
        public int bossId;
        /// <summary>
        /// 所属boss 的关键key
        /// </summary>
        public string BossKey => $"{zoneid}-{bossId}";

        /// <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 void Close()
        {
            this.CurrentState = ClientState.Leaved;                              // 关闭接收窗口
            Room.Instance.RemovePeer(this);                                      // 从房间中移除自己
            //SaveDamage();
            this.sendDataBuffer.Reader.Completion
                .ContinueWith(t =>
                Task.Delay(5000).ContinueWith(t1 => Sock.Close())
                );                                                               // 延迟5秒关闭套接字
        }
        public void AddDamage(int damage)
        {
            if (Properties.ContainsKey(PropertyName.TotalDamage))
            {
                var totalDamage = Convert.ToInt32(Properties[PropertyName.TotalDamage]);
                Properties[PropertyName.TotalDamage] = damage + totalDamage;
            }
            else
            {
                Properties.Add(PropertyName.TotalDamage, damage);
            }
            SaveDamage(damage);
        }
        void SaveDamage(int damage)
        {
            try
            {
                var rdb = Redis.Ins.GetDatabase(0);
                if (Properties.TryGetValue(PropertyName.Name, out var name)
                    && Properties.TryGetValue(PropertyName.Zoneid, out var zoneid))
                {
                    var roomName = Room.Instance.Name ?? "Room";
                    var redis_key = MemKey_Game.BossFight_Damage_byDateHour_zset(Convert.ToInt32(zoneid), bossId, roomName);
                    var who = UID + "__" + name.ToString();
                    rdb.SortedSetIncrement(redis_key, who, damage);
                }
                else
                {
                    Console.WriteLine("uid/zoneid 为空.");
                }
            }
            catch (RedisTimeoutException te)
            {
                Console.WriteLine("redis操作超时");
            }
        }

        /// <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))                         //取出一条完整数据
                    {
                        await Room.Instance.MsgChannel.Writer.WriteAsync(new KeyValuePair<int, sSocketData>(this.Id, _socketData));    // 放入channel
                    }
                    Array.Clear(arrServerRecMsg, 0, length);
                }
                catch (SocketException e)
                {
                    if (e.ErrorCode == 10060)                                          // 超时的时候错误号码是10060
                    {
                        continue;                                                      // 继续等待  
                    }
                    break;
                }
                catch (Exception)
                {
                    break;
                }
            }

            Close();                                                                 // 关闭
        }

    }
}