WEP_V2.1.3/SourceCode/We.App/WeApp.Core/CommonManager/MqttClient/WeAppMsgClientManager.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;

using Abp.Configuration;
using Abp.Dependency;
using Abp.Domain.Repositories;
using Abp.Domain.Uow;
using Abp.Runtime.Caching;

using IwbZero.ToolCommon;
using IwbZero.ToolCommon.LogHelpers;
using IwbZero.ToolCommon.StringModel;

using Microsoft.AspNet.SignalR;

using uPLibrary.Networking.M2Mqtt.Messages;

using WeApp.Configuration;
using WeApp.Configuration.Cache;
using WeApp.Hubs;
using WeApp.TrainingCamp;

using WeEngine.CommonDto;
using WeEngine.Enum;
using WeEngine.Message;
using WeEngine.Packages;

using WeMessageService;

namespace WeApp.CommonManager.MqttClient
{
    public class WeAppMsgClientManager : ISingletonDependency
    {
        public WeAppMsgClientManager(ISettingManager settingManager, IRepository<CampGroupInfo, string> groupRepository,
            IRepository<GroupScoreInfo> gsRepository, IRepository<CampInfo, string> campRepository, IUnitOfWorkManager unitOfWorkManager, ICacheManager cacheManager, IRepository<GroupLogInfo> glRepository)
        {
            SettingManager = settingManager;
            GroupRepository = groupRepository;
            GsRepository = gsRepository;
            CampRepository = campRepository;
            UnitOfWorkManager = unitOfWorkManager;
            CacheManager = cacheManager;
            GlRepository = glRepository;
            IwbHubContext = GlobalHost.ConnectionManager.GetHubContext<IwbHub>();
            if (!Client.IsConnected)
            {
                RegisterClient();
            }
        }

        private static readonly Dictionary<string, ScenePusher> _scenePushes = new Dictionary<string, ScenePusher>();
        private string _url => SettingManager.GetSettingValue(IwbSettingNames.WeDataCenterIp);
        private string _engineUrl => SettingManager.GetSettingValue(IwbSettingNames.WeEngineIp);
        private static M2MqttMsgClient Client = new M2MqttMsgClient();
        private IUnitOfWorkManager UnitOfWorkManager { get; }
        private ISettingManager SettingManager { get; }
        private ICacheManager CacheManager { get; }
        private IRepository<CampInfo, string> CampRepository { get; }
        private IRepository<CampGroupInfo, string> GroupRepository { get; }
        private IRepository<GroupScoreInfo> GsRepository { get; }
        private IRepository<GroupLogInfo> GlRepository { get; }
        protected IHubContext IwbHubContext { get; }

        #region MSG-Client

        public string ClientId { get; set; }

        /// <summary>
        /// 强制刷新客户端
        /// </summary>
        public void RefreshClient()
        {
            this.LogInfo($"客户端{ClientId}开始强制刷新！");
            Client.DisConnect();
            Client = new M2MqttMsgClient();
            RegisterClient();
            SendMessage(ClientId, $"客户端{ClientId}已强制刷新！");
            SendMessage("WeEngine", $"客户端{ClientId}已强制刷新！");
        }

        /// <summary>
        /// 注册客户端
        /// </summary>
        public void RegisterClient()
        {
            ClientId = IwbConsts.ClientName;
#if DEBUG
            var result = new MsgClientDto()
            {
                AppId = ClientId,
                Name = "智慧演练系统",
                SecretKey = "WePlatformWeApp",
                ServerIp = "localhost",
                ServerPort = 1885,
            };
#else
            MsgClientDto result = CacheManager.GetMsgClientDto(ClientId, _url);
            if (result == null)
            {
                //throw new UserFriendlyException("未查询到消息客户端配置！");
                this.LogError($"未查询到消息客户端{ClientId}的配置！");
                return;
            }
#endif

            Client.SetOptions(result.ServerIp, result.ServerPort, ClientId, result.Name, result.SecretKey);
            Client.SetReceivedHandler(ReceivedHandler);
            Subscribe();
        }

        private void Subscribe()
        {
            if (Client.CheckClient())
            {
                Client.SubscribeTopic(ClientId, false);
                IwbHubContext.Clients.Group(ClientId).getMqttMsg(new { topic = ClientId, msg = "订阅主题" });
                Client.SubscribeTopic("Heartbeat");
                IwbHubContext.Clients.Group(ClientId).getMqttMsg(new { topic = $"{ClientId}/Heartbeat", msg = "订阅主题" });
                var t = typeof(MessageType);
                var fts = t.GetFields(BindingFlags.Static | BindingFlags.Public);
                foreach (var f in fts)
                {
                    string name = f.Name;
                    Client.SubscribeTopic($"{name}/+");
                    IwbHubContext.Clients.Group(ClientId).getMqttMsg(new { topic = $"{ClientId}/{name}/+", msg = "订阅主题" });
                    //AsyncHelper.RunSync(() => Client.SubscribeTopic(name));
                }
            }
        }

        public void SendMessage(WeMessage message, MessageType type, string clientId = null)
        {
            if (Client.NotValidate)
            {
                RegisterClient();
            }

            var msg = message.Obj2String();
            Client.SendMsg(type.ToString(), msg, clientId);
            IwbHubContext.Clients.Group(ClientId).getSendMqttMsg(new { topic = $"{ClientId}/{type}/", msg });
        }

        public void SendMessage(WeMessage message)
        {
            if (Client.NotValidate)
            {
                RegisterClient();
            }

            var msg = message.Obj2String();
            Client.SendMsg(message.Type.ToString(), msg, message.TargetClientId);
            IwbHubContext.Clients.Group(ClientId).getSendMqttMsg(new { topic = $"{message.TargetClientId}/{message.Type}/", msg });
        }

        public void SendMessage(string msg)
        {
            if (Client.NotValidate)
            {
                RegisterClient();
            }
            Client.SendMessage(ClientId, msg);
            IwbHubContext.Clients.Group(ClientId).getSendMqttMsg(new { topic = $"{ClientId}", msg });
        }

        public void SendMessage(string topic, string msg)
        {
            if (Client.NotValidate)
            {
                RegisterClient();
            }
            Client.SendMessage(topic, msg);
            IwbHubContext.Clients.Group(ClientId).getSendMqttMsg(new { topic, msg });
        }

        /// <summary>
        /// 消息接受回调处理
        /// </summary>
        /// <param name="msg"></param>
        private void ReceiveMsgHandler(string msg)
        {
            try
            {
                var message = msg.Str2Obj<WeMessage>();
                if (message != null)
                {
                    switch (message.Type)
                    {
                        case MessageType.Start:
                            ReceiveStart(message);
                            break;

                        case MessageType.End:
                            ReceiveEnd(message);
                            break;

                        case MessageType.RunningInfo:
                            ReceiveRunningInfo(message);
                            break;

                        case MessageType.RoundScore:
                            ReceiveRoundScore(message);
                            break;

                        case MessageType.SceneInfo:
                            ReceiveScene(message);
                            break;

                        case MessageType.GuideInfo:
                            ReceiveGuide(message);
                            break;

                        case MessageType.Environment:
                            ReceiveEnvironment(message);
                            break;

                        case MessageType.CommandMatchSuccess:
                            ReceiveMatchSuccess(message);
                            break;

                        case MessageType.CommonMessage:
                            ReceiveCommonMsg(message);
                            break;
                    }
                }
                else
                {
                    typeof(WeAppMsgClientManager).LogError("MSG ERROR:" + msg);
                }
            }
            catch (Exception e)
            {
                typeof(WeAppMsgClientManager).LogError(e);
            }
        }

        /// <summary>
        /// 消息接受回调处理
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void ReceivedHandler(object sender, MqttMsgPublishEventArgs e)
        {
            string msg = System.Text.Encoding.UTF8.GetString(e.Message);
            IwbHubContext.Clients.Group(ClientId).getMqttMsg(new { topic = e.Topic, msg });
            if (e.Topic == ClientId || e.Topic == $"{ClientId}/Heartbeat")
            {
                return;
            }
            ReceiveMsgHandler(msg);
        }

        //private IMqttApplicationMessageReceivedHandler ReceivedHandler =>
        //    new MqttApplicationMessageReceivedHandlerDelegate(
        //         (arg) =>
        //        {
        //            string payload = System.Text.Encoding.UTF8.GetString(arg.ApplicationMessage.Payload);
        //            var message = payload.Str2Obj<WeMessage>();
        //            if (message != null)
        //            {
        //                switch (message.MsgType)
        //                {
        //                    case MessageType.Environment:
        //                        break;

        //                    case MessageType.RunningInfo:
        //                         ReceiveRunningInfo(message);
        //                        break;

        //                    case MessageType.RoundScore:
        //                         ReceiveRoundScore(message);
        //                        break;

        //                    case MessageType.CommonMessage:
        //                        break;

        //                    case MessageType.GuideInfo:
        //                        break;
        //                }
        //            }
        //            else
        //            {
        //                typeof(WeAppMsgClientManager).LogError("Message ERROR:" + payload);
        //            }
        //        });

        //public static readonly List<string> StartGroupNos = new List<string>();

        #endregion MSG-Client

        [UnitOfWork]
        protected virtual void ReceiveStart(WeMessage message)
        {
            var groupNo = message.RunningId;
            var group = GroupRepository.GetAllIncluding(a => a.CampInfo).FirstOrDefault(a => a.Id == groupNo);
            if (group != null)
            {
                var date = DateTime.Now;

                group.CampGroupState = CampGroupStateDefinition.Run;
                group.StartDate = date;
                GroupRepository.Update(group);
                UnitOfWorkManager.Current.SaveChanges();
                if (group.CampInfo.CampState != CampStateDefinition.Run)
                {
                    group.CampInfo.CampState = CampStateDefinition.Run;
                    group.CampInfo.StartDate = date;
                    CampRepository.Update(group.CampInfo);
                    UnitOfWorkManager.Current.SaveChanges();
                    IwbHubContext.Clients.Group(group.CampNo).getReloadSpec(new { group.CampNo }.Obj2String());
                }
                ReloadPage(group.CampNo, groupNo);
                ScenePusher scenePusher = new ScenePusher();
                scenePusher.SendMsg += (msg) => { PushMessage(msg, groupNo); };
                _scenePushes[groupNo] = scenePusher;
            }
        }

        /// <summary>
        /// 刷新前台页面
        /// </summary>
        /// <param name="campNo"></param>
        /// <param name="no"></param>
        private void ReloadPage(string campNo, string no)
        {
            IwbHubContext.Clients.Group(campNo).getReloadStu(new { no }.Obj2String());
            IwbHubContext.Clients.Group(campNo).getReloadLeader(new { no }.Obj2String());
            IwbHubContext.Clients.Group(campNo).getReloadPublic(new { no }.Obj2String());
        }

        [UnitOfWork]
        protected virtual void ReceiveEnd(WeMessage message)
        {
            var groupNo = message.RunningId;
            //var group = GroupRepository.GetAllIncluding(a=>a.CampInfo).FirstOrDefault(a => a.Id == groupNo);
            //if (group != null)
            //{
            //    var date = DateTime.Now;
            //    this.LogDebug($"培训营分组【{groupNo}】结束");

            //    group.CampGroupState = CampGroupStateDefinition.End;
            //    group.EngDate = date;
            //    group.TrainingMinute = Convert.ToDateTime(group.StartDate).GetTimeSpanMinute(date);
            //    GroupRepository.Update(group);
            //    UnitOfWorkManager.Current.SaveChanges();
            //    var count = GroupRepository.Count(a => a.CampGroupState != CampGroupStateDefinition.End && a.CampNo == group.CampNo);
            //    if (count  == 0)
            //    {
            //        this.LogDebug($"培训营【{group.CampNo}】结束");
            //        var camp = CampRepository.FirstOrDefault(a => a.Id == group.CampNo);
            //        if (camp != null)
            //        {
            //            camp.CampState = CampStateDefinition.End;
            //            camp.EngDate = date;
            //            camp.TrainingMinute = Convert.ToDateTime(group.CampInfo.StartDate).GetTimeSpanMinute(date);
            //            CampRepository.Update(camp);
            //            UnitOfWorkManager.Current.SaveChanges();
            //        }
            //    }
            //    StartGroupNos.Remove(groupNo);

            //}
            StopPushScene(groupNo);
        }

        [UnitOfWork]
        protected virtual void ReceiveRunningInfo(WeMessage message)
        {
            var groupNo = message.RunningId;
            // var runningInfo = message.Content.Str2Obj<RunningInfo>();
            var runningInfo = GetRunningInfo(groupNo);
            if (runningInfo != null)
            {
                var group = GroupRepository.FirstOrDefault(a => a.Id == groupNo);
                if (group != null)
                {
                    if (group.RoundIndex != runningInfo.CurrentRoundIndex)
                    {
                        IwbHubContext.Clients.Group(group.CampNo).getReloadLeader(new { no = groupNo }.Obj2String());
                    }
                    runningInfo.Name = group.Name;
                    CacheManager.GetCache(IwbCacheNames.GroupRunningInfoCache).Set(groupNo, runningInfo);
                    group.RunningInfo = runningInfo.Obj2String();
                    group.RoundIndex = runningInfo.CurrentRoundIndex;
                    GroupRepository.Update(group);
                    UnitOfWorkManager.Current.SaveChanges();
                    IwbHubContext.Clients.Group(group.CampNo).getRunningChange(new { no = groupNo }.Obj2String());
                }
            }
        }

        private PackageDataNode GetRunningInfo(string no)
        {
            var url =
                $"{_engineUrl.Ew("/")}api/services/Engine/Run/QueryPackageDataNode?id={no}";
            var result = url.RequestPost("");
            var info = result.Str2Obj<PackageDataNode>();
            return info;
        }

        [UnitOfWork]
        protected virtual void ReceiveRoundScore(WeMessage message)
        {
            var groupNo = message.RunningId;
            var roundScoreInfo = message.Content.Str2Obj<RoundScoreInfo>();
            if (roundScoreInfo == null)
            {
                return;
            }
            var group = GroupRepository.FirstOrDefault(a => a.Id == groupNo);
            if (group == null)
            {
                return;
            }
            //var gs = await GsRepository.FirstOrDefaultAsync(a =>
            //    a.GroupNo == groupNo && a.RoundIndex == roundScoreInfo.RoundIndex);
            var gs = CacheManager.GetGroupRoundScore(groupNo, roundScoreInfo.RoundIndex, GsRepository).Result;
            List<TagScoreInfo> tagScores = new List<TagScoreInfo>(), pushScores = new List<TagScoreInfo>(),
                oldTagScores = gs?.BehaviorTagScoreInfo.Str2Obj<List<TagScoreInfo>>();
            var activeTag = CacheManager.GetCampTags(group.CampNo, CampRepository).Result;
            if (roundScoreInfo.TagScores != null)
            {
                foreach (var tNo in activeTag)
                {
                    var tagScore = roundScoreInfo.TagScores.FirstOrDefault(a => a.TagNo.Trim() == tNo.Trim());
                    if (tagScore != null)
                    {
                        pushScores.Add(new TagScoreInfo()
                        {
                            TagNo = tNo,
                            SystemScore = tagScore.SystemScore,
                            TotalScore = tagScore.TotalScore
                        });
                        var oldTagScore = oldTagScores?.FirstOrDefault(a => a.TagNo.Trim() == tagScore.TagNo.Trim());
                        if (oldTagScore != null)
                        {
                            tagScore.SystemScore = oldTagScore.SystemScore + tagScore.SystemScore;
                            tagScore.CorrectionScore = oldTagScore.CorrectionScore == 0
                                ? 0
                                : oldTagScore.CorrectionScore + tagScore.SystemScore;
                            tagScore.SystemScore = tagScore.SystemScore > tagScore.TotalScore
                                ? tagScore.TotalScore
                                : tagScore.SystemScore;
                        }
                    }
                    else
                    {
                        tagScore = oldTagScores?.FirstOrDefault(a => a.TagNo.Trim() == tNo.Trim()) ?? new TagScoreInfo()
                        {
                            TagNo = tNo,
                            CorrectionScore = 0,
                            SystemScore = 0,
                            TotalScore = 0
                        };
                    }

                    tagScores.Add(tagScore);
                }
            }

            if (roundScoreInfo.KeyWords.IsNotEmpty())
            {
                var arr = roundScoreInfo.KeyWords.Split('|');
                if (int.TryParse(arr[0], out var logId))
                {
                    var log = GlRepository.FirstOrDefault(a => a.Id == logId);
                    if (log != null)
                    {
                        log.LogState = LogStateDefinition.HasMatch;
                        GlRepository.Update(log);
                    }
                }
            }
            if (gs == null)
            {
                gs = new GroupScoreInfo()
                {
                    CampNo = group.CampNo,
                    GroupNo = groupNo,
                    RoundIndex = roundScoreInfo.RoundIndex,
                    SystemScore = roundScoreInfo.RoundScore,
                    BehaviorTagScoreInfo = tagScores.Obj2StringCamelCase()
                };
                GsRepository.Insert(gs);
            }
            else
            {
                gs.CampGroupInfo = null;
                gs.SystemScore = roundScoreInfo.RoundScore;
                gs.BehaviorTagScoreInfo = tagScores.Obj2StringCamelCase();
                GsRepository.Update(gs);
            }

            CacheManager.GetCache(IwbCacheNames.GroupRunningInfoCache)
                .Set($"SCORE-{groupNo}-{roundScoreInfo.RoundIndex}", gs);
            IwbHubContext.Clients.Group(group.CampNo)
                .getRoundScore(new
                {
                    no = groupNo,
                    round = roundScoreInfo.RoundIndex,
                    systemScore = roundScoreInfo.RoundScore,
                    tagScores = pushScores
                }.Obj2StringCamelCase());
        }

        [UnitOfWork]
        protected virtual void ReceiveScene(WeMessage message)
        {
            var groupNo = message.RunningId;
            var scene = message.Content.Str2Obj<SceneDto>();
            if (scene != null)
            {
                var group = GroupRepository.FirstOrDefault(a => a.Id == groupNo);
                if (group != null)
                {
                    var sceneNoDto = CacheManager.GetSceneNos(groupNo, GroupRepository).Result;
                    sceneNoDto.hasChange = true;
                    sceneNoDto.PathList = sceneNoDto.PathList ?? new List<string>();
                    if (!sceneNoDto.PathList.Contains(scene.Path))
                    {
                        sceneNoDto.PathList.Add(scene.Path);
                    }
                    CacheManager.GetCache(IwbCacheNames.SceneInfoCache).Set($"SceneNos-{groupNo}", sceneNoDto);

                    //IwbHubContext.Clients.Group(group.CampNo).getNewScene(new { no = groupNo, sceneNo = scene.No }.Obj2String());
                    //ScenePusher scenePusher;
                    //if (_scenePushes.ContainsKey(groupNo))
                    //{
                    //    scenePusher = _scenePushes[groupNo];
                    //}
                    //else
                    //{
                    //    scenePusher = new ScenePusher();
                    //    scenePusher.SendMsg += (msg) => { PushMessage(msg, groupNo); };
                    //}
                    //scenePusher.Push(new ScenePushMessage(groupNo, scene.Path));
                    //_scenePushes[groupNo] = scenePusher;
                    SendScenePusher(groupNo, groupNo, new ScenePushMessage(groupNo, scene.Path));
                }
            }
        }

        [UnitOfWork]
        protected virtual void ReceiveGuide(WeMessage message)
        {
            var groupNo = message.RunningId;
            var group = GroupRepository.FirstOrDefault(a => a.Id == groupNo);
            if (group == null)
            {
                return;
            }

            //ScenePusher scenePusher;
            //if (_scenePushes.ContainsKey(groupNo))
            //{
            //    scenePusher = _scenePushes[groupNo];
            //}
            //else
            //{
            //    scenePusher = new ScenePusher();
            //    _scenePushes[groupNo] = scenePusher;
            //    scenePusher.SendMsg += (msg) => { PushMessage(msg, groupNo); };
            //}
            //scenePusher.Push(new ScenePushMessage(PushMessageDefinition.Guide) { content = message.Content });
            //IwbHubContext.Clients.Group(groupNo).getGuide(message.Content);
            SendScenePusher(groupNo, groupNo, new ScenePushMessage(PushMessageDefinition.Guide) { content = message.Content });
        }

        [UnitOfWork]
        protected virtual void ReceiveEnvironment(WeMessage message)
        {
            var groupNo = message.RunningId;
            var group = GroupRepository.FirstOrDefault(a => a.Id == groupNo);
            if (group == null)
            {
                return;
            }

            //ScenePusher scenePusher;
            //if (_scenePushes.ContainsKey(groupNo))
            //{
            //    scenePusher = _scenePushes[groupNo];
            //}
            //else
            //{
            //    scenePusher = new ScenePusher();
            //    _scenePushes[groupNo] = scenePusher;
            //    scenePusher.SendMsg += (msg) => { PushMessage(msg, groupNo); };
            //}
            //scenePusher.Push(new ScenePushMessage(PushMessageDefinition.Environment) { content = message.Content });
            SendScenePusher(groupNo, groupNo, new ScenePushMessage(PushMessageDefinition.Environment) { content = message.Content });
        }

        [UnitOfWork]
        protected virtual void ReceiveMatchSuccess(WeMessage message)
        {
            var groupNo = message.RunningId;
            var group = GroupRepository.FirstOrDefault(a => a.Id == groupNo);
            if (group == null)
            {
                return;
            }
            IwbHubContext.Clients.Group(groupNo).getMatchSuccess(message.Content);
        }

        [UnitOfWork]
        protected virtual void ReceiveCommonMsg(WeMessage message)
        {
            var groupNo = message.RunningId;
            var group = GroupRepository.FirstOrDefault(a => a.Id == groupNo);
            if (group == null)
            {
                return;
            }

            //ScenePusher scenePusher;
            //if (_scenePushes.ContainsKey(groupNo))
            //{
            //    scenePusher = _scenePushes[groupNo];
            //}
            //else
            //{
            //    scenePusher = new ScenePusher();
            //    _scenePushes[groupNo] = scenePusher;
            //    scenePusher.SendMsg += (msg) => { PushMessage(msg, groupNo); };
            //}
            //scenePusher.Push(new ScenePushMessage(PushMessageDefinition.CommonMsg) { content = message.Content });
            SendScenePusher(groupNo, groupNo, new ScenePushMessage(PushMessageDefinition.CommonMsg) { content = message.Content });
        }

        /// <summary>
        /// 开始推送情景（解决页面还未加载时就推送情景）
        /// 需要页面第一次加载打开推送开关后再开始推送
        /// </summary>
        /// <param name="groupNo"></param>
        public static bool StartPushScene(string groupNo)
        {
            ScenePusher scenePusher = _scenePushes.ContainsKey(groupNo) ? _scenePushes[groupNo] : new ScenePusher();
            _scenePushes[groupNo] = scenePusher;

            return scenePusher.Start();
        }

        public static void StopPushScene(string no)
        {
            if (_scenePushes.ContainsKey(no))
            {
                _scenePushes[no].Stop();
            }
        }

        private void SendScenePusher(string no, string code, ScenePushMessage pushMsg)
        {
            ScenePusher scenePusher;

            if (_scenePushes.ContainsKey(no))
            {
                scenePusher = _scenePushes[no];
                if (scenePusher.NeedHandel)
                {
                    scenePusher.SendMsg += (msg) => { PushMessage(msg, code); };
                }
            }
            else
            {
                scenePusher = new ScenePusher();
                scenePusher.SendMsg += (msg) => { PushMessage(msg, code); };
                _scenePushes[no] = scenePusher;
            }

            scenePusher.Push(pushMsg);
        }

        private void PushMessage(ScenePushMessage msg, string code)
        {
            switch (msg.type)
            {
                case PushMessageDefinition.Scene:
                    IwbHubContext.Clients.Group(code).getNewScene(msg);
                    break;

                case PushMessageDefinition.Guide:
                    IwbHubContext.Clients.Group(code).getGuide(msg.content);
                    break;

                case PushMessageDefinition.Environment:
                    IwbHubContext.Clients.Group(code).getEnvironment(msg.content);
                    break;

                case PushMessageDefinition.CommonMsg:
                    IwbHubContext.Clients.Group(code).getCommonMsg(msg.content);
                    break;
            }
        }
    }
}