using CCrazyStudio.Core.Common.Protocols;
using CrazyStudio.Core.Common.Redis;
using CrazyStudio.Core.Common.Redis.Dto;
using CrazyStudio.Core.Common.Status.Eitities;
using CrazyStudio.Core.Common.Status.Interfaces;
using CrazyStudio.Core.Common.Tools;
using Fleck;
using MessagePack;
using Newtonsoft.Json.Linq;
using Org.BouncyCastle.Bcpg;
using StackExchange.Redis;
using System;
using System.Collections.Concurrent;
using System.Diagnostics.PerformanceData;
using System.Net.Sockets;
using System.Threading.Tasks;
using static System.Net.Mime.MediaTypeNames;
using static System.Runtime.InteropServices.JavaScript.JSType;

namespace CrazyStudio.Core.Common.Status
{
    /// <summary>
    /// 分布式集群用户状态，消息，订阅管理类
    /// </summary>
    public class ClusterOnlineUsers : IOnlineUser
    {
        /// <summary>
        /// 组用户信息，一对多关系，组名对应多个用户
        /// </summary>
        public ConcurrentDictionary<string, ConcurrentDictionary<string, OnlineUserInfo>> GroupRooms { get; set; } = new ConcurrentDictionary<string, ConcurrentDictionary<string, OnlineUserInfo>>();

        /// <summary>
        /// 链接ID对应的用户信息，一对一
        /// </summary>
        public ConcurrentDictionary<string, OnlineUserInfo> SocketUsers { get; set; } = new ConcurrentDictionary<string, OnlineUserInfo>();

        /// <summary>
        /// 本地房间信息，一对一
        /// </summary>
        public ConcurrentDictionary<string, RoomInfo> Rooms { get; set; } = new ConcurrentDictionary<string, RoomInfo>();

        private RedisHelper _redisHelper;

        private string _myServerClusterTag = string.Empty;


        public ClusterOnlineUsers(RedisHelper redisHelper, string serverClusterTag)
        {
            this._redisHelper = redisHelper;
            redisHelper.GetSubscribeEvent += RedisHelper_GetSubscribeEvent;
            redisHelper.SubscribeSystemChannels().Wait();
            _myServerClusterTag = serverClusterTag;
        }


        private async Task<bool> HandleClusterSubscribe(ClusterMessage? clusterMessage)
        {
            if (clusterMessage != null)
            {
                if (clusterMessage?.MsgType == XNetEventKeys.CLUSTER_JOIN)
                {
                    Console.ForegroundColor = ConsoleColor.Green;
                    string clusterTag = Encoding.UTF8.GetString(clusterMessage?.MsgContent!);

                    if (clusterTag == _myServerClusterTag)
                    {
                        Console.WriteLine($"集群本节点服务器频道 {clusterTag} 订阅成功");
                        Console.ForegroundColor = ConsoleColor.White;
                    }
                    else
                    {
                        Console.WriteLine($"集群其他节点服务器频道 {clusterTag} 订阅成功");
                        Console.ForegroundColor = ConsoleColor.White;
                        await _redisHelper.PublishAsync(clusterTag, new ClusterMessage
                        {
                            MsgType = XNetEventKeys.CLUSTER_JOIN_REPLY,
                            MsgContent = Encoding.UTF8.GetBytes(_myServerClusterTag)
                        });
                    }
                }
                else if (clusterMessage?.MsgType == XNetEventKeys.CLUSTER_JOIN_REPLY)
                {
                    Console.ForegroundColor = ConsoleColor.Green;
                    string clusterTag = Encoding.UTF8.GetString(clusterMessage?.MsgContent!);
                    Console.WriteLine($"集群其他节点服务器频道 {clusterTag} 订阅成功");
                    Console.ForegroundColor = ConsoleColor.White;
                }
                return true;
            }
            return false;
        }

        private async void RedisHelper_GetSubscribeEvent(RedisChannel channel, RedisValue data)
        {
            var clusterMessage = MessagePackSerializer.Deserialize<ClusterMessage>(data);
            if (!await HandleClusterSubscribe(clusterMessage))
            {
                await SendMessageToConnection(clusterMessage);
            }
        }

        private async Task RedisHelperGetUserSubscribeHandler(string connectionId, ClusterMessage clusterMessage)
        {
            if (!clusterMessage.ExceptIds.Contains(connectionId))
            {
                clusterMessage.ConnectionId = connectionId;
                await SendMessageToConnection(clusterMessage);
            }
        }

        private void AddOrUpdateUserToGroup(OnlineUserInfo user)
        {
            var users = GroupRooms.AddOrUpdate(user.RoomId, (key) =>
            {
                var thisGroupUsers = new ConcurrentDictionary<string, OnlineUserInfo>();
                thisGroupUsers[user.ConnectionId] = user;
                return thisGroupUsers;
            }, (key, oldValue) =>
            {
                oldValue[user.ConnectionId] = user;
                return oldValue;
            });
        }

        /// <summary>
        /// 添加或者更新用户信息
        /// </summary>
        /// <param name="connectionId">链接ID，唯一</param>
        /// <param name="user">在线用户信息类</param>
        public async Task AddOrUpdateUser(string connectionId, OnlineUserInfo user)
        {
            OnlineUserInfo? existUser = null;
            if (!SocketUsers.TryGetValue(connectionId, out existUser))
            {
                existUser = user;
                SocketUsers.TryAdd(connectionId, existUser);
            }

            if (existUser != null && !string.IsNullOrEmpty(existUser.UserName))
            {
                string prefixUserName = PrefixTool.GetRedisUserPatchID(existUser.UserName);
                OnlineUserInfo? redisUser = await _redisHelper.GetObjAsync<OnlineUserInfo>(prefixUserName).ConfigureAwait(false);

                var getSubscribeEvent = new Action<RedisChannel, RedisValue>(async (channel, value) =>
                {
                    var clusterMessage = MessagePackSerializer.Deserialize<ClusterMessage>(value);
                    await RedisHelperGetUserSubscribeHandler(connectionId, clusterMessage).ConfigureAwait(false);
                });

                Action<RedisChannel, RedisValue>? getGroupSubscribeEvent = null;

                bool isJoinRoom = !string.IsNullOrEmpty(user.RoomId);
                if (isJoinRoom)
                {
                    getGroupSubscribeEvent = new Action<RedisChannel, RedisValue>(async (channel, value) =>
                    {
                        if (GroupRooms.TryGetValue(user.RoomId, out var users))
                        {
                            var clusterMessage = MessagePackSerializer.Deserialize<ClusterMessage>(value);
                            foreach (var user in users)
                            {
                                await RedisHelperGetUserSubscribeHandler(user.Value.ConnectionId, clusterMessage).ConfigureAwait(false);
                            }
                            if (clusterMessage.MsgType == XNetEventKeys.GAME_END)
                            {
                                GroupRooms.TryRemove(user.RoomId, out var _);
                                Rooms.TryRemove(user.RoomId, out var _);
                            }
                        }
                    });
                }
                if (redisUser == null)
                {
                    redisUser = existUser;

                    //redisUser.AtServers = new ConcurrentDictionary<string, string>();
                    //redisUser.AtServers[connectionId] = _myServerClusterTag;
                    redisUser.ConnectionId = connectionId;
                    redisUser.RoomId = user.RoomId;
                    redisUser.AvatarUrl = user.AvatarUrl;

                    await _redisHelper.SubscribeAsync(redisUser.UserName, getSubscribeEvent).ConfigureAwait(false);
                    if (getGroupSubscribeEvent != null)
                    {
                        await _redisHelper.SubscribeAsync(redisUser.RoomId, getGroupSubscribeEvent).ConfigureAwait(false);
                    }

                    await _redisHelper.SetObjAsync(prefixUserName, redisUser).ConfigureAwait(false);
                }
                else
                {
                    //if (redisUser.AtServers.Any())
                    //{
                    //    redisUser.AtServers[connectionId] = _myServerClusterTag;
                    //}
                    //else
                    //{
                    //    redisUser.AtServers = new ConcurrentDictionary<string, string>();
                    //    redisUser.AtServers[connectionId] = _myServerClusterTag;
                    //}

                    //if (SocketUsers.TryGetValue(redisUser.ConnectionId, out var oldUser))
                    //{
                    //    if (oldUser.Socket != user.Socket)
                    //    {
                    //        if (oldUser.ConnectionId != user.ConnectionId)
                    //        {
                    //            await RemoveUserFromId(oldUser.ConnectionId);
                    //        }
                    //    }
                    //}


                    await _redisHelper.UnsubscribeAsync(redisUser.UserName).ConfigureAwait(false);


                    redisUser.UserName = user.UserName;
                    redisUser.AvatarUrl = user.AvatarUrl;
                    redisUser.ConnectionId = user.ConnectionId;
                    redisUser.RoomId = user.RoomId;

                    await _redisHelper.SubscribeAsync(redisUser.UserName, getSubscribeEvent).ConfigureAwait(false);
                    if (getGroupSubscribeEvent != null)
                    {
                        await _redisHelper.SubscribeAsync(redisUser.RoomId, getGroupSubscribeEvent).ConfigureAwait(false);
                    }

                    await _redisHelper.SetObjAsync(prefixUserName, redisUser).ConfigureAwait(false);
                }

                AddOrUpdateUserToGroup(user);

                await UpdateRoom(redisUser, isJoinRoom).ConfigureAwait(false);
            }
        }

        /// <summary>
        /// 创建或者更新房间
        /// </summary>
        /// <param name="redisUser"></param>
        /// <returns></returns>
        private async Task UpdateRoom(OnlineUserInfo redisUser, bool isIncrementUserCount)
        {
            if (!string.IsNullOrEmpty(redisUser.RoomId))
            {
                string prefixRoomId = PrefixTool.GetRedisRoomPatchID(redisUser.RoomId);
                string roomLockKey = Global.GetIdLockKey(redisUser.RoomId);
                await _redisHelper.LockTakeAsync(roomLockKey, Global.ServerToken);
                //获取游戏房间，不存在则创建
                var roomInfo = await _redisHelper.GetObjAsync<RoomInfo>(prefixRoomId).ConfigureAwait(false);
                var sendHostMessage = MessagePackSerializer.Serialize(new BaseMessage
                {
                    ConnectionId = redisUser.ConnectionId,
                    SenderId = redisUser.UserName,
                    MsgType = XNetEventKeys.HOST,
                    ReceiverOrRoomId = redisUser.UserName
                });
                if (roomInfo == null)
                {
                    isIncrementUserCount = true;
                    redisUser.IsHost = true;
                    roomInfo = new RoomInfo
                    {
                        RoomId = redisUser.RoomId,
                        UserNameList = new HashSet<string>() { redisUser.UserName },
                        //RoomName = redisUser.RoomId,
                        HostUserName = redisUser.UserName
                        //RoomUserCount = 1
                    };
                    await _redisHelper.SetObjAsync(prefixRoomId, roomInfo, TimeSpan.FromSeconds(Global.RoomExpirySecond)).ConfigureAwait(false);
                    await SendMessageToGroupOrUser(redisUser.UserName, sendHostMessage).ConfigureAwait(false);
                }
                else
                {
                    //Console.WriteLine("roominfo is exist");
                    if (isIncrementUserCount)
                    {
                        bool isSendHost = false;
                        //Console.WriteLine("isIncrementUserCount");
                        if (!roomInfo.UserNameList.Contains(redisUser.UserName))
                        {
                            //Console.WriteLine("Contains");
                            roomInfo.UserNameList.Add(redisUser.UserName);
                            //Interlocked.Increment(ref roomInfo.RoomUserCount);

                            if (string.IsNullOrEmpty(roomInfo.HostUserName))
                            {
                                redisUser.IsHost = true;
                                roomInfo.HostUserName = redisUser.UserName;
                                if (roomInfo.UserNameList.Count == 1)
                                {
                                    isSendHost = true;
                                    await SendMessageToGroupOrUser(redisUser.UserName, sendHostMessage).ConfigureAwait(false);
                                }
                            }
                        }
                        //Console.WriteLine("roomInfo.UserNameList.Count " + roomInfo.UserNameList.Count);
                        if (!isSendHost && roomInfo.UserNameList.Count == 1)
                        {
                            await SendMessageToGroupOrUser(redisUser.UserName, sendHostMessage).ConfigureAwait(false);
                        }
                    }
                }

                if (isIncrementUserCount)
                {
                    if (roomInfo != null)
                    {
                        if (!Rooms.TryGetValue(roomInfo.RoomId, out var localRoom))
                        {
                            localRoom = roomInfo;
                            Rooms.TryAdd(roomInfo.RoomId, localRoom);
                        }


                        if (SocketUsers.TryGetValue(redisUser.ConnectionId, out var existUsers))
                        {
                            localRoom.StartServer(async () =>
                            {
                                await SyncRoomMessage(localRoom).ConfigureAwait(false);

                            }, Global.FlightTime * 1000, async () =>
                            {
                                existUsers.HeartBeatDurationMs = -1;//游戏房间服务停止，修正心跳检测为不需要
                                localRoom?.Dispose();
                                //await SyncRoomMessage(localRoom!).ConfigureAwait(false);//停止转发服务之前发送完剩下的消息
                                await SendGameEndMessage(localRoom!).ConfigureAwait(false);
                            });
                        }


                        await _redisHelper.SetObjAsync(prefixRoomId, roomInfo, TimeSpan.FromSeconds(Global.RoomExpirySecond)).ConfigureAwait(false);
                    }
                }
                else
                {
                    if (roomInfo != null)
                    {
                        await _redisHelper.SetObjAsync(prefixRoomId, roomInfo, TimeSpan.FromSeconds(Global.RoomExpirySecond)).ConfigureAwait(false);
                    }
                }
                await _redisHelper.LockReleaseAsync(roomLockKey, Global.ServerToken);
            }
        }

        private async Task SyncRoomMessage(RoomInfo roomInfo)
        {
            if (roomInfo != null)
            {
                List<byte[]> msgs = roomInfo.PopRoomMessageThisTime();
                for (int i = 0; i < msgs.Count; i++)
                {
                    await _redisHelper.PublishAsync(roomInfo.RoomId, new ClusterMessage
                    {
                        MsgContent = msgs[i]
                    }).ConfigureAwait(false);
                }
            }
        }

        private async Task SendGameEndMessage(RoomInfo roomInfo)
        {
            if (roomInfo != null)
            {
                ClusterMessage? msg = new ClusterMessage
                {
                    MsgType = XNetEventKeys.GAME_END,
                    MsgContent = MessagePackSerializer.Serialize(new BaseMessage { MsgType = XNetEventKeys.GAME_END })
                };
                await _redisHelper.PublishAsync(roomInfo.RoomId, msg).ConfigureAwait(false);
            }
        }

        /// <summary>
        /// 根据链接ID移除集合相关用户信息
        /// </summary>
        /// <param name="connectionId">链接ID</param>
        public async Task RemoveUserFromId(string connectionId)
        {
            if (SocketUsers.TryRemove(connectionId, out var removeUser))
            {
                if (!string.IsNullOrEmpty(removeUser.UserName))
                {
                    string prefixUserName = PrefixTool.GetRedisUserPatchID(removeUser.UserName);
                    var redisUser = await _redisHelper.GetObjAsync<OnlineUserInfo>(prefixUserName).ConfigureAwait(false);

                    if (redisUser != null)
                    {
                        await _redisHelper.UnsubscribeAsync(redisUser.UserName).ConfigureAwait(false);

                        await RemoveRedisUser(redisUser, removeUser);
                    }
                    if (removeUser.Socket != null && removeUser.Socket.IsAvailable)
                    {
                        removeUser.Socket.Close();
                    }
                }
            }
        }

        private async Task RemoveRedisUser(OnlineUserInfo redisUser, OnlineUserInfo removeUser)
        {
            await RemoveUserFromGroup(redisUser.RoomId, removeUser);

            string prefixUserName = PrefixTool.GetRedisUserPatchID(redisUser.UserName);
            await _redisHelper.KeyDeleteAsync(prefixUserName).ConfigureAwait(false);


            string prefixRoomId = PrefixTool.GetRedisRoomPatchID(redisUser.RoomId);
            //获取游戏房间
            var roomInfo = await _redisHelper.GetObjAsync<RoomInfo>(prefixRoomId).ConfigureAwait(false);

            if (roomInfo != null)
            {
                roomInfo.MessageQueue.TryRemove(removeUser.UserName, out var _);
                roomInfo.UserNameList.Remove(removeUser.UserName);
                //Interlocked.Decrement(ref roomInfo.RoomUserCount);
                //如果房间人数大于0，则往房间用户发送当前用户离线消息
                if (roomInfo.UserNameList.Count > 0)
                {
                    if (redisUser.IsHost)
                    {
                        roomInfo.HostUserName = string.Empty;
                    }
                    await _redisHelper.SetObjAsync(prefixRoomId, roomInfo, TimeSpan.FromSeconds(Global.RoomExpirySecond)).ConfigureAwait(false);
                    await SendMessageToGroupOrUser(redisUser.RoomId, MessagePackSerializer.Serialize(new BaseMessage
                    {
                        MsgType = XNetEventKeys.OFFLINE,
                        ReceiverOrRoomId = redisUser.UserName,
                    }));

                    await RequestHost(redisUser);
                }
                else
                {
                    //Rooms.TryRemove(redisUser.RoomId, out var localRoom);
                    //if (localRoom != null)
                    //{
                    //    localRoom.Dispose();
                    //}
                    //房间人数为0，删除房间
                    await _redisHelper.KeyDeleteAsync(prefixRoomId).ConfigureAwait(false);
                }
            }
        }

        /// <summary>
        /// 这里在房间用户离线的时候需触发
        /// </summary>
        /// <param name="userName"></param>
        /// <returns></returns>
        public async Task RequestHost(OnlineUserInfo user, bool isRemoveExist = true)
        {
            string prefixRoomId = PrefixTool.GetRedisRoomPatchID(user.RoomId);
            var roomInfo = await _redisHelper.GetObjAsync<RoomInfo>(prefixRoomId).ConfigureAwait(false);

            if (roomInfo != null)
            {
                if (isRemoveExist)
                {
                    roomInfo.UserNameList.Remove(user.UserName);
                }
                if (roomInfo.UserNameList.Count > 0)
                {
                    var hostUserName = roomInfo.UserNameList.FirstOrDefault();
                    if (!string.IsNullOrEmpty(hostUserName))
                    {
                        roomInfo.HostUserName = hostUserName;
                        string prefixUserName = PrefixTool.GetRedisUserPatchID(hostUserName);
                        string roomLockKey = Global.GetIdLockKey(roomInfo.RoomId);
                        var redisUser = await _redisHelper.GetObjAsync<OnlineUserInfo>(prefixUserName).ConfigureAwait(false);
                        if (redisUser != null)
                        {
                            redisUser.IsHost = true;

                            //Console.WriteLine(hostUserName);
                            await _redisHelper.SetObjAsync(prefixRoomId, roomInfo, TimeSpan.FromSeconds(Global.RoomExpirySecond)).ConfigureAwait(false);
                            await _redisHelper.SetObjAsync(prefixUserName, redisUser).ConfigureAwait(false);

                            await SendMessageToGroupOrUser(redisUser.UserName, MessagePackSerializer.Serialize(new BaseMessage
                            {
                                ConnectionId = redisUser.ConnectionId,
                                SenderId = redisUser.UserName,
                                MsgType = XNetEventKeys.HOST,
                                ReceiverOrRoomId = redisUser.UserName
                            })).ConfigureAwait(false);
                        }
                    }
                }
            }
        }

        /// <summary>
        /// 根据用户名移除集合相关用户信息
        /// </summary>
        /// <param name="connectionId">链接ID</param>
        public async Task RemoveUserFromName(string userName)
        {
            if (!string.IsNullOrEmpty(userName))
            {
                string prefixUserName = PrefixTool.GetRedisUserPatchID(userName);

                var redisUser = await _redisHelper.GetObjAsync<OnlineUserInfo>(prefixUserName).ConfigureAwait(false);

                if (redisUser != null)
                {
                    if (SocketUsers.TryRemove(redisUser.ConnectionId, out var user))
                    {
                        await RemoveUserFromGroup(user.RoomId, user);
                        if (user.Socket != null && user.Socket.IsAvailable)
                        {
                            user.Socket.Close();
                        }
                    }

                    await RemoveRedisUser(redisUser, user!);
                }
            }
        }

        private async Task RemoveUserFromGroup(string groupId, OnlineUserInfo user)
        {
            if (groupId != null && GroupRooms.TryGetValue(groupId, out var users))
            {
                if (user != null)
                {
                    users.TryRemove(user.ConnectionId, out var _);
                }
                if (!users.Any())
                {
                    GroupRooms.TryRemove(groupId, out var _);
                    await _redisHelper.UnsubscribeAsync(groupId).ConfigureAwait(false);
                }
            }
        }

        ///// <summary>
        ///// 根据用户名获取对应用户名在线用户(包括WebSocketConnect 对象)
        ///// </summary>
        ///// <param name="userName"></param>
        ///// <returns>对应用户名在线用户信息，找不到就返回空值</returns>
        //public OnlineUserInfo? GetUser(string userName)
        //{
        //    if (!string.IsNullOrEmpty(userName))
        //    {
        //        var onlineUser = SocketUsers.Where(s => s.Value.UserName == userName).FirstOrDefault().Value;
        //        return onlineUser;
        //    }
        //    return null;
        //}

        /// <summary>
        /// 根据用户名获取对应用户名Redis在线用户信息
        /// </summary>
        /// <param name="userName"></param>
        /// <returns>对应用户名在线用户信息，找不到就返回空值</returns>
        public async Task<OnlineUserInfo?> GetRedisUser(string userName)
        {
            if (!string.IsNullOrEmpty(userName))
            {
                string prefixUserName = PrefixTool.GetRedisUserPatchID(userName);
                var redisUser = await _redisHelper.GetObjAsync<OnlineUserInfo>(prefixUserName).ConfigureAwait(false);
                return redisUser;
            }
            return null;
        }

        /// <summary>
        /// 根据链接ID获取对应用户名在线用户
        /// </summary>
        /// <param name="connectionId">链接ID</param>
        /// <returns>查询到的在线用户信息，找不到就返回空值</returns>
        public OnlineUserInfo? GetUsersFromConnectionId(string connectionId)
        {
            if (!string.IsNullOrEmpty(connectionId))
            {
                if (SocketUsers.TryGetValue(connectionId, out var user))
                {
                    return user;
                }
            }
            return null;
        }



        ///// <summary>
        ///// 发送数据给用户名对应的所有链接
        ///// </summary>
        ///// <param name="userName">用户名</param>
        ///// <param name="packet">要发送的字节数组</param>
        ///// <returns></returns>
        //public async Task SendMessageToUser(string userName, byte[] packet)
        //{
        //    if (!string.IsNullOrEmpty(userName))
        //    {
        //        var redisUser = await GetRedisUser(userName).ConfigureAwait(false);
        //        if (redisUser != null)
        //        {
        //            await _redisHelper.PublishAsync(userName, new ClusterMessage
        //            {
        //                ConnectionId = redisUser.ConnectionId,
        //                Data = packet
        //            });
        //        }
        //    }
        //}


        /// <summary>
        /// 发送数据给链接ID对应的链接
        /// </summary>
        /// <param name="userName">链接ID</param>
        /// <param name="packet">要发送的字节数组</param>
        /// <returns></returns>
        public async Task<bool> SendMessageToConnection(string connectionId, byte[] packet)
        {
            if (!string.IsNullOrEmpty(connectionId))
            {
                if (SocketUsers.TryGetValue(connectionId, out var user))
                {
                    if (user != null)
                    {
                        await Send(user, packet);
                    }
                }
            }
            return false;
        }


        /// <summary>
        /// 发送数据给链接ID对应的链接,本地没有走集群
        /// </summary>
        /// <param name="userName">链接ID</param>
        /// <param name="packet">要发送的字节数组</param>
        /// <returns></returns>
        public async Task SendMessageToUser(string connectionId, string userName, byte[] packet)
        {
            if (!await SendMessageToConnection(connectionId, packet))
            {
                await _redisHelper.PublishAsync(userName, new ClusterMessage
                {
                    ConnectionId = connectionId,
                    MsgContent = packet
                });
            }
        }

        /// <summary>
        /// 发送数据给链接对应用户的链接,本地没有走集群
        /// </summary>
        /// <param name="onlineUser">在线用户对象</param>
        /// <param name="packet">要发送的字节数组</param>
        /// <returns></returns>
        public async Task SendMessageToUser(OnlineUserInfo onlineUser, byte[] packet)
        {
            if (!await SendMessageToConnection(onlineUser.ConnectionId, packet))
            {
                await _redisHelper.PublishAsync(onlineUser.UserName, new ClusterMessage
                {
                    ConnectionId = onlineUser.ConnectionId,
                    MsgContent = packet
                });
            }
        }

        /// <summary>
        /// 发送集群数据给链接ID对应的链接
        /// </summary>
        /// <param name="packet">接收到，要发送的集群数据</param>
        /// <returns></returns>
        public async Task SendMessageToConnection(ClusterMessage clusterData)
        {
            if (!string.IsNullOrEmpty(clusterData.ConnectionId))
            {
                if (SocketUsers.TryGetValue(clusterData.ConnectionId, out var user))
                {
                    if (user != null)
                    {
                        await Send(user, clusterData.MsgContent!);
                    }
                }
            }
        }

        private async Task Send(OnlineUserInfo user, byte[] data)
        {
            try
            {
                if (user.Socket.IsAvailable)
                {
                    await user.Socket.Send(data);// clusterData.MsgContent);
                }
            }
            catch (IOException)
            {
            }
            catch (SocketException)
            {
            }
            catch (ConnectionNotAvailableException)
            {

            }
            catch (AggregateException)
            {

            }
        }

        public async Task PushMessageToRoomQueue(string groupOrUserName, string userName, byte[] packet)
        {
            if (Rooms.TryGetValue(groupOrUserName, out var room))
            {
                if (room.RoomServer != null && room.RoomServer!.IsRunning)
                {
                    room.PushMessage(userName, packet);
                }
                else
                {
                    await SendMessageToGroupOrUser(groupOrUserName, packet);
                }
            }
            else
            {
                await SendMessageToGroupOrUser(groupOrUserName, packet);
            }
        }

        /// <summary>
        /// 发送数据给组ID对应的链接
        /// </summary>
        /// <param name="groupOrUserName">组ID或者用户名</param>
        /// <param name="packet">要发送的字节数组</param>
        /// <returns></returns>
        public async Task SendMessageToGroupOrUser(string groupOrUserName, byte[] packet)
        {
            await _redisHelper.PublishAsync(groupOrUserName, new ClusterMessage
            {
                MsgContent = packet
            });
        }

        public async Task SendMessageToGroupOrUserExceptIds(string groupOrUserName, byte[] packet, HashSet<string> exceptIds)
        {
            await _redisHelper.PublishAsync(groupOrUserName, new ClusterMessage
            {
                MsgContent = packet,
                ExceptIds = exceptIds
            });
        }
    }
}
