using POSV.OtherWaiMai; using POSV.Proxy.ThirdPartyOrder; using System; using System.Collections.Generic; using System.Linq; using System.Text; using uPLibrary.Networking.M2Mqtt; using uPLibrary.Networking.M2Mqtt.Messages; namespace POSV.Proxy.Mqtt { public class MqttClientUtils { private static object _lock = new object(); private static NLog.Logger logger = NLog.LogManager.GetCurrentClassLogger(); private static MqttClientUtils _instance = null; private MqttClient mqttClient = null; private System.Timers.Timer _reconnectTimer; private Action _action = null; public static MqttClientUtils Instance { get { if (_instance == null) { lock (_lock) { _instance = new MqttClientUtils(); } } return _instance; } } private bool isRunning = false; private void Reconnect() { //Global.Instance.Worker.StoreInfo.Id = "t_AGNTBICRy0 "; bool isException = false; try { if (isRunning) { logger.Info("连接mqtt正在运行,忽略本次执行"); isException = true; return; } isRunning = true; logger.Info("尝试连接外卖消息中心....."); //2022-12-13 11:06 zhangy Edit 更换IP地址为域名 this.mqttClient = new MqttClient("mqtt.ffcygl.com", 18830, false, null, null, MqttSslProtocols.None); //this.mqttClient = new MqttClient("127.0.0.1", 1883, false, null, null, MqttSslProtocols.None); this.mqttClient.ConnectionClosed -= OnMqttConnectionClosed; this.mqttClient.ConnectionClosed += OnMqttConnectionClosed; this.mqttClient.MqttMsgSubscribed -= OnMqttMsgSubscribed; this.mqttClient.MqttMsgSubscribed += OnMqttMsgSubscribed; this.mqttClient.MqttMsgPublishReceived -= OnMqttMsgPublishReceived; this.mqttClient.MqttMsgPublishReceived += OnMqttMsgPublishReceived; string clientId = string.Format("{0}-{1}-{2}-{3}", "cy2", Global.Instance.Worker.TenantId, Global.Instance.Worker.StoreInfo.Id, Global.Instance.Authc.PosNo); //默认的用户名和密码连接,为MQTT消息中心启用密码访问准备 this.mqttClient.Connect(clientId, Global.Instance.Authc.PosNo, Global.Instance.Authc.PosId, false, 30); } catch (Exception ex) { isException = true; if (this.IsMqttConnected) { this.mqttClient.Unsubscribe(this.Topics.Item1); this.mqttClient.Disconnect(); } this.mqttClient = null; logger.Error(ex, "外卖消息中心连接异常"); } finally { //没有异常发生 if (!isException) { logger.Info("连接外卖消息中心成功....."); this.mqttClient.Subscribe(this.Topics.Item1, this.Topics.Item2); } isRunning = false; } } private void OnMqttMsgSubscribed(object sender, MqttMsgSubscribedEventArgs e) { logger.Info("订阅外卖消息中心<{0}>...", e.MessageId); } private void OnMqttConnectionClosed(object sender, EventArgs e) { logger.Info("断开外卖消息中心连接..."); } public bool IsMqttConnected { get { return this.mqttClient != null && this.mqttClient.IsConnected; } } private void ReconnectCallback(object sender, System.Timers.ElapsedEventArgs e) { if (this.IsMqttConnected) { logger.Info("外卖消息中心连接状态:<{0}>,{1}", this.mqttClient.IsConnected, this.mqttClient.ClientId); } else { this.Reconnect(); } } /// /// 处理收到的消息 /// /// public void ReceiveMessage(Action action) { this._action = action; } /// /// 外卖接单被pengjd调整到这里作为入口。 /// zhangy 2020-02-19 Add 添加注释 /// /// /// void OnMqttMsgPublishReceived(object sender, MqttMsgPublishEventArgs e) { string message = Encoding.UTF8.GetString(e.Message); string topic = e.Topic; logger.Info("服务端推送<{0}>消息:{1}",topic, message); if (topic.StartsWith("program")) { logger.Info("收到巨为小程序订单"); JuWeiAppOrderProxy.sendJwWeixinMessage(message); OnMessageReceived?.Invoke(sender, new MessageEventArgs(0, message)); } if (topic.StartsWith("waimai")) { //开启商家版美团外卖和餐道外卖 var enableShopMeiTuan = Global.Instance.GlobalConfigBoolValue(ConfigConstant.CONFIG_CASHIER_ENABLESHOPMEITUAN,false); if (enableShopMeiTuan) { logger.Info("收到美团商家外卖订单"); OtherWaiMaiApi.sendSJWaiMaiMessage(message); } else { logger.Info("收到外卖订单"); OtherWaiMaiApi.sendWaiMaiMessage(message); } OnMessageReceived?.Invoke(sender, new MessageEventArgs(0, message)); } if (topic.StartsWith("weixin")) { logger.Info("微信小程序订单"); OtherWaiMaiApi.sendWeixinMessage(message); OnMessageReceived?.Invoke(sender, new MessageEventArgs(1, message)); } if (topic.StartsWith("qimai")) { logger.Info("收到企迈第三方订单"); OtherWaiMaiApi.sendQiMaiMessage(message); OnMessageReceived?.Invoke(sender, new MessageEventArgs(1, message)); } this._action?.Invoke(e); } public ushort Subscribe(string topics) { return this.mqttClient.Subscribe(new string[] { topics }, new byte[] { MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE }); } public ushort Subscribe(string[] topics, byte[] levels) { return this.mqttClient.Subscribe(topics, levels); } public ushort Publish(string topic, string message) { return this.mqttClient.Publish(topic, Encoding.UTF8.GetBytes(message), MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, false); } public delegate void EventHandler(object sender, MessageEventArgs e); public event EventHandler OnMessageReceived; private Tuple Topics { get { //运维参数 string tenant = string.Format("{0}/{1}/{2}", "devops", "cy2", Global.Instance.Worker.TenantId); string shop = string.Format("{0}/{1}/{2}/{3}", "devops", "cy2", Global.Instance.Worker.TenantId, Global.Instance.Worker.StoreInfo.No); string pos = string.Format("{0}/{1}/{2}/{3}/{4}", "devops", "cy2", Global.Instance.Worker.TenantId, Global.Instance.Worker.StoreInfo.No,Global.Instance.Authc.PosNo); //业务参数 string waimai = string.Format("{0}/{1}/{2}/{3}", "waimai", "cy2", Global.Instance.Worker.TenantId, Global.Instance.Worker.StoreInfo.Id); string weixin = string.Format("{0}/{1}/{2}/{3}", "weixin", "cy2", Global.Instance.Worker.TenantId, Global.Instance.Worker.StoreInfo.Id); string qimai = string.Format("{0}/{1}/{2}/{3}", "qimai", "cy2", Global.Instance.Worker.TenantId, Global.Instance.Worker.StoreInfo.Id); string program = string.Format("{0}/{1}/{2}/{3}", "program", "cy2", Global.Instance.Worker.TenantId, Global.Instance.Worker.StoreInfo.Id); string[] topics = { waimai, weixin, qimai, program , tenant ,shop , pos}; byte[] levels = { MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE }; return new Tuple(topics, levels); } } /// /// 启动任务 /// public void Startup() { try { if (this._reconnectTimer == null) { this._reconnectTimer = new System.Timers.Timer(); } //设置定时间隔(毫秒为单位) this._reconnectTimer.Interval = 55000; //设置执行一次(false)还是一直执行(true) this._reconnectTimer.AutoReset = true; //设置是否执行System.Timers.Timer.Elapsed事件 this._reconnectTimer.Enabled = true; //绑定Elapsed事件 this._reconnectTimer.Elapsed += ReconnectCallback; logger.Info("初始化外卖消息中心......"); } catch (Exception ex) { logger.Info(ex, "外卖消息中心启动异常"); } finally { Reconnect(); } } /// /// 停止任务 /// public void Stop() { if (this.IsMqttConnected) { //取消订阅 this.mqttClient.Unsubscribe(this.Topics.Item1); //断开连接 this.mqttClient.Disconnect(); //关闭时间 this._reconnectTimer.Stop(); } } } public class MessageEventArgs : EventArgs { private readonly int _type; private readonly string _message; public MessageEventArgs(int type, string message) { this._type = type; this._message = message; } public string Message { get { return _message; } } public int Type { get { return _type; } } } }