WEP_V2.1.3/SourceCode/We.Message/M2MqttMsgClient/M2MqttMsgClient.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using IwbZero.ToolCommon.LogHelpers;
using IwbZero.ToolCommon.StringModel;
using uPLibrary.Networking.M2Mqtt;
using uPLibrary.Networking.M2Mqtt.Exceptions;
using uPLibrary.Networking.M2Mqtt.Messages;
using Timer = System.Timers.Timer;

//using uPLibrary.Networking.M2Mqtt;
//using uPLibrary.Networking.M2Mqtt.Messages;

namespace WeMessageService
{
    public class M2MqttMsgClient
    {
        public M2MqttMsgClient()
        {
            TryReconnectCount = 0;
            ToClose = false;
            TopicList = new List<string>();
            HeartbeatTopic = new List<string>();
            HeartbeatSwitch = false;
        }
        private MqttClient Client { get; set; }
        private string ServerIp { get; set; }
        private int ServerPort { get; set; }
        private string ClientId { get; set; }
        private string ServerUserId { get; set; }
        private string ServerPassword { get; set; }
        private List<string> TopicList { get; }
        private byte[] QosLevels { get; set; } = { MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE };
        private int TryReconnectCount { get; set; }
        private bool ToClose { get; set; }
        private bool HeartbeatSwitch { get; set; }
        private Timer Timer { get; set; }
        public MqttClient.MqttMsgPublishEventHandler MsgReceivedHandler { get; set; }
        public bool NotValidate => (ServerIp.IsEmpty() || ServerPort == 0 || ServerIp.IsEmpty() ||
                                    ClientId.IsEmpty() || ServerUserId.IsEmpty() || ServerPassword.IsEmpty());

        public List<string> HeartbeatTopic { get; set; }
        public bool IsConnected => Client?.IsConnected ?? false;

        /// <summary>
        /// 配置参数
        /// </summary>
        /// <param name="serverIp"></param>
        /// <param name="serverPort"></param>
        /// <param name="clientId"></param>
        /// <param name="serverUserId"></param>
        /// <param name="serverPassword"></param>
        public void SetOptions(string serverIp, int serverPort, string clientId, string serverUserId, string serverPassword)
        {
            ServerIp = serverIp;
            ServerPort = serverPort;
            ClientId = clientId;
            ServerUserId = serverUserId;
            ServerPassword = serverPassword;
            ToClose = false;
            TryReconnectCount = 0;
        }

        /// <summary>
        /// 配置消息接受回调事件
        /// </summary>
        /// <param name="handler"></param>
        public void SetReceivedHandler(MqttClient.MqttMsgPublishEventHandler handler)
        {
            MsgReceivedHandler = handler;
        }

        /// <summary>
        /// 配置消息质量
        /// </summary>
        /// <param name="qosLevels"></param>
        public void SetQosLevels(byte[] qosLevels)
        {
            QosLevels = qosLevels;
        }

        /// <summary>
        /// 订阅主题
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="withClient"></param>
        public void SubscribeTopic(string topic,bool withClient=true)
        {
            if (CheckClient())
            {
                topic = withClient ? $"{ClientId}/{topic}" : $"{topic}";
                if (!TopicList.Contains(topic))
                {
                    TopicList.Add(topic);
                }
                Client.Subscribe(new[] { topic }, QosLevels);
                this.LogInfo("SUB-TOPIC:" + topic);
            }
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="msg"></param>
        /// <param name="clientId"></param>
        public void SendMsg(string topic, string msg, string clientId = null)
        {
            if (CheckClient())
            {
                var msgContent = Encoding.UTF8.GetBytes(msg);
                clientId = clientId ?? ClientId;
                var targetClientIds = clientId.Split('@');
                foreach (var targetClientId in targetClientIds)
                {
                   var topicTemp = $"{targetClientId}/{topic}";
                    Client.Publish(topicTemp, msgContent);
                }
                this.LogInfo($"【SEND-MSG:[{clientId}] [{topic}][{(msg.Length > 100 ? msg.Substring(0, 100) : msg)}]");
            }
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="msg"></param>
        public void SendMessage(string topic, string msg)
        {
            if (CheckClient())
            {
                Client.Publish(topic, Encoding.UTF8.GetBytes(msg));
                this.LogInfo($"【SEND-MSG:[{topic}][{(msg.Length>100? msg.Substring(0,100):msg)}]");
            }
           
        }

        /// <summary>
        /// 断开连接
        /// </summary>
        public void DisConnect()
        {
            if (IsConnected)
            {
                ToClose = true;
                try
                {
                    Client.Disconnect();
                }
                catch (MqttClientException e)
                {
                    this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
                }
                catch (MqttCommunicationException e)
                {
                    this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
                }
                catch (MqttConnectionException e)
                {
                    this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
                }
                catch (MqttTimeoutException e)
                {
                    this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
                }
                catch (Exception e)
                {
                    this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
                }

            }
        }

        /// <summary>
        /// 检查客户端是否连接
        /// </summary>
        /// <returns></returns>
        public bool CheckClient()
        {
            if (IsConnected)
                return true;

            if (NotValidate)
            {
                //throw new UserFriendlyException("客户端配置参数不合法，请检查后再试！");
                this.LogError($"客户端{ClientId}配置参数[{ServerIp}:{ServerPort}/{ServerUserId}/{ServerPassword}]不合法，请检查后再试！");
                return false;
            }
                
            if (Client == null)
            {
                Client = CreateClient();
                Connect();
            }
            else if (!IsConnected)
            {
                Connect();
            }
            if (!IsConnected)
            {
                //throw new UserFriendlyException("消息客户端未连接，请检查！");
                this.LogError("消息客户端未连接，请检查！");
                return false;
            }

            return true;
        }

        /// <summary>
        /// 创建客户端
        /// </summary>
        /// <returns></returns>
        private MqttClient CreateClient()
        {
            if (NotValidate)
            {
                //throw new UserFriendlyException("客户端配置参数不合法，请检查后再试！");
                this.LogError($"客户端{ClientId}配置参数[{ServerIp}:{ServerPort}/{ServerUserId}/{ServerPassword}]不合法，请检查后再试！");
                return null;
            }

            try
            {
                ToClose = false;
                TryReconnectCount = 0;
                var client = new MqttClient(ServerIp, ServerPort, false, null, null, MqttSslProtocols.None);
                client.ConnectionClosed += ClientClosed;
                //client.MqttMsgSubscribed += MsgSubscribed;
                client.MqttMsgUnsubscribed += MsgUnSubscribed;
                client.MqttMsgPublished += MsgPublished;
                client.MqttMsgPublishReceived += MsgPublishReceived;
                if (MsgReceivedHandler != null)
                {
                    client.MqttMsgPublishReceived += MsgReceivedHandler;
                }
                return client;
            }
            catch(MqttClientException e)
            {
                this.LogError("客户端创建失败");
                this.LogError(e);
                //throw new UserFriendlyException("客户端创建失败，请检查后再试！");

            }
            catch(Exception e)
            {
                this.LogError("客户端创建失败");
                this.LogError(e);
                //throw new UserFriendlyException("客户端创建失败，请检查后再试！");
            }

            return null;
        }


        private void Connect()
        {
            if (NotValidate)
            {
                //throw new UserFriendlyException("客户端配置参数不合法，请检查后再试！");
                this.LogError($"客户端{ClientId}配置参数[{ServerIp}:{ServerPort}/{ServerUserId}/{ServerPassword}]不合法，请检查后再试！");
                return;
            }
            if (ToClose)
                return;
            try
            {
                Client?.Connect(ClientId, ServerUserId, ServerPassword);
            }
            catch (MqttClientException e)
            {
                this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
            }
            catch (MqttCommunicationException e)
            {
                this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
            }
            catch (MqttConnectionException e)
            {
                this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
            }
            catch (MqttTimeoutException e)
            {
                this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
            }
            catch (Exception e)
            {
                this.LogError($"Connect Exception:[{TryReconnectCount}]-[{e.Message}]");
            }
            if (IsConnected)
            {
                ReSubscribeTopic();
                this.LogInfo(TryReconnectCount > 0
                    ? $"客户端{ClientId}已重新注册连接！重连次数：{TryReconnectCount}"
                    : $"客户端{ClientId}已注册连接！");
                TryReconnectCount = 0;
                if (HeartbeatSwitch)
                {
                    if (Timer != null)
                    {
                        Timer.Stop();
                        Timer.Close();
                        Timer = null;
                    }
                    Timer = new Timer(1000 * 60 * 5) { AutoReset = true, Enabled = true };
                    Timer.Elapsed += (o, e) =>
                    {

                        SendMsg("Heartbeat", $"HB-{new Random().Next(11111, 99999)}");
                        if (HeartbeatTopic.Any())
                        {
                            foreach (var topic in HeartbeatTopic)
                            {
                                SendMessage($"{topic}/Heartbeat", $"[{ClientId}]：HB-{new Random().Next(11111, 99999)}");
                            }
                        }
                    };
                }
            }
            else
            {
                if (TryReconnectCount <= 10)
                {
                    Thread.Sleep(5000);
                    TryContinueConnect();
                }
                else
                {
                    try
                    {
                        Client?.Disconnect();
                    }
                    catch (Exception e)
                    {
                        this.LogError(e);
                    }
                    
                }
            }
        }

        public void ReSubscribeTopic(List<string> topicList = null)
        {
            topicList = topicList ?? TopicList;
            if (topicList.Any() && IsConnected) 
            {
                foreach (var topic in TopicList)
                {
                    Client?.Subscribe(new[] { topic }, QosLevels);
                }
            }
        }
        private void TryContinueConnect()
        {
            if (IsConnected) return;
            if (ToClose) return;
            if (Client == null)
            {
                CreateClient();
                Thread.Sleep(3000);
            }
            TryReconnectCount++;
            if (TryReconnectCount > 10)
            {
                ToClose = true;
            }
            Connect();
        }

        #region EVENT

        private void ClientClosed(object sender, EventArgs e)
        {
            this.LogInfo($"Connection Lost:{ClientId}");
            TryContinueConnect();
        }

        //private void MsgSubscribed(object sender, MqttMsgSubscribedEventArgs e)
        //{
        //    this.LogInfo($"Subscribed:{e.MessageId}");
        //}
        private void MsgUnSubscribed(object sender, MqttMsgUnsubscribedEventArgs e)
        {
            this.LogInfo($"UnSubscribed:{e.MessageId}");
        }

        private void MsgPublished(object sender, MqttMsgPublishedEventArgs e)
        {
            this.LogInfo($"Published:{e.MessageId}-{e.IsPublished}");
        }
        // 处理接收到的消息
        private void MsgPublishReceived(object sender, MqttMsgPublishEventArgs e)
        {
            string msg = Encoding.UTF8.GetString(e.Message);
            this.LogInfo($"RECEIVED-MSG】: [{e.Topic}] [{(msg.Length > 500 ? msg.Substring(0, 500) : msg)}]");
        } 
        #endregion
    }
}