//using BPA.MQTTClient; //using Microsoft.Extensions.Configuration; //using MQTTnet.Client; //using MQTTnet.Client; //using MQTTnet; //using MQTTnet.Client; //using MQTTnet.Client.Options; //using MQTTnet.Client.Receiving; using BPASmartClient.Helper; using BPASmartClient.Message; using MQTTnet; using MQTTnet.Client; using MQTTnet.Client.Options; using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; namespace BPASmartClient.MQTT { public class MQTTProxy { private IMqttClient client; IMqttClientOptions options; public Action MessageRecive { get; set; } public Action Connected { get; set; } public Action LostConnect { get; set; } public bool IsConnected { get; set; } bool push = true; Action UseDisconnectedAction; public async void Connect(string UserName, string pass, string IP, int port, string clientID) { options = new MqttClientOptionsBuilder().WithTcpServer(IP, port).WithClientId(clientID).WithCredentials(UserName, pass).Build(); client = new MqttFactory().CreateMqttClient(); client.UseDisconnectedHandler(c => { if (UseDisconnectedAction == null) { Reconnect();//注册UseDisconnectedAction委托 UseDisconnectedAction();//执行委托 } }).UseApplicationMessageReceivedHandler(c => { MessageRecive?.Invoke(Encoding.UTF8.GetString(c.ApplicationMessage.Payload)); }).UseConnectedHandler((e) => { //MessageLog.GetInstance.Show($"连接成功"); }); try { await client.ConnectAsync(options); } catch (Exception ex) { MessageLog.GetInstance.ShowEx(ex.Message); MessageLog.GetInstance.Show("mqtt连接失败!重连执行中"); } if (client.IsConnected) { MessageLog.GetInstance.Show("MQTT连接成功!"); Connected?.Invoke(); } } private void Reconnect() { UseDisconnectedAction = new Action(() => { MessageLog.GetInstance.ShowEx("MQTT 断开连接"); Thread.Sleep(2000); while (!UniversalHelper.GetInstance().GetNetworkState()) { Thread.Sleep(2000); } bool ErrorFlag = false; while (!client.IsConnected) { try { MessageLog.GetInstance.Show($"重连中"); client.ConnectAsync(options).Wait(); } catch (Exception ex) { if (!ErrorFlag) { MessageLog.GetInstance.ShowEx(ex.ToString()); ErrorFlag = true; } } Thread.Sleep(3000); } if (client.IsConnected) { MessageLog.GetInstance.Show("MQTT重连成功!"); LostConnect?.Invoke(); } UseDisconnectedAction = null; }); } /// /// Mqtt 订阅 /// /// 需要订阅的主题 public async void MqttSubscriptionAsync(string topic) { if (client != null && client.IsConnected) { try { var result = await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).WithExactlyOnceQoS().Build()); } catch { } } } /// /// Mqtt 订阅 /// /// 需要订阅的主题 public async void Subscrib(params string[] topic) { if (client != null && client.IsConnected) { try { foreach (var item in topic) { var result = await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(item).WithExactlyOnceQoS().Build()); } } catch { } } } /// /// Mqtt 发布 /// /// 需要发布的主题 /// 需要发布的内容 public async void Publish(string topic, string content) { ; if (client != null && client.IsConnected) { push = true; var msg = new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(content).WithExactlyOnceQoS().Build(); try { var result = await client.PublishAsync(msg); } catch { } } if(!client.IsConnected) { if(push) { MessageLog.GetInstance.ShowEx("MQTT未连接"); push = false; } } } //public Action MessageRecive { get; set; } //public Action Connected { get; set; } //public Action LostConnect { get; set; } //public bool IsConnected { get; set; } //private IMqttClient client; //public void Connect(string userName, string Password, string ip, int port, string clientId) //{ // IConfigurationBuilder configurationBuilder = new ConfigurationBuilder(); // configurationBuilder.AddMqttClientHostedService(p => // { // p.Server = ip; // p.Port = port; // //p.UserName = "rafiul"; // //p.Password = "12345678"; // p.UserName = userName; // p.Password = Password; // p.mqttClientConnectedHandlerDelegate = new MQTTnet.Client.Connecting.MqttClientConnectedHandlerDelegate(e => // { // IsConnected = true; // Connected?.Invoke(); // }); // //p.mqttClientDisconnectedHandlerDelegate = new MQTTnet.Client.Disconnecting.MqttClientDisconnectedHandlerDelegate(e => // //{ // // IsConnected = false; // // LostConnect?.Invoke(); // //}); // p.ConnectedResult += (s, e) => // { // client = e; // }; // p.MqttApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e => // { // MessageRecive?.Invoke(Encoding.Default.GetString(e.ApplicationMessage.Payload)); // }); // }); //} public void CloseConnect() { client.Dispose(); } //public async void Publish(string topic, string content) //{ // if (client.IsConnected) // await client.PublishAsync(topic, content); //} //public async void Subscrib(params string[] topics) //{ // foreach (var topic in topics) // { // await client.SubscribeAsync(new MqttTopicFilter() { Topic = topic, QualityOfServiceLevel = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce }); // } //} } }