终端一体化运控平台
Ви не можете вибрати більше 25 тем Теми мають розпочинатися з літери або цифри, можуть містити дефіси (-) і не повинні перевищувати 35 символів.
 
 
 

272 рядки
7.8 KiB

  1. //using BPA.MQTTClient;
  2. //using Microsoft.Extensions.Configuration;
  3. //using MQTTnet.Client;
  4. //using MQTTnet.Client;
  5. //using MQTTnet;
  6. //using MQTTnet.Client;
  7. //using MQTTnet.Client.Options;
  8. //using MQTTnet.Client.Receiving;
  9. using BPASmartClient.Helper;
  10. using BPASmartClient.Message;
  11. using MQTTnet;
  12. using MQTTnet.Client;
  13. using MQTTnet.Client.Options;
  14. using System;
  15. using System.Collections.Generic;
  16. using System.IO;
  17. using System.Linq;
  18. using System.Text;
  19. using System.Threading;
  20. using System.Threading.Tasks;
  21. namespace BPASmartClient.MQTT
  22. {
  23. public class MQTTProxy
  24. {
  25. private IMqttClient client;
  26. IMqttClientOptions options;
  27. public Action<string> MessageRecive { get; set; }
  28. public Action Connected { get; set; }
  29. public Action LostConnect { get; set; }
  30. public bool IsConnected { get; set; }
  31. bool push = true;
  32. Action UseDisconnectedAction;
  33. public async void Connect(string UserName, string pass, string IP, int port, string clientID)
  34. {
  35. options = new MqttClientOptionsBuilder().WithTcpServer(IP, port).WithClientId(clientID).WithCredentials(UserName, pass).Build();
  36. client = new MqttFactory().CreateMqttClient();
  37. client.UseDisconnectedHandler(c =>
  38. {
  39. if (UseDisconnectedAction == null)
  40. {
  41. Reconnect();//注册UseDisconnectedAction委托
  42. UseDisconnectedAction();//执行委托
  43. }
  44. }).UseApplicationMessageReceivedHandler(c =>
  45. {
  46. MessageRecive?.Invoke(Encoding.UTF8.GetString(c.ApplicationMessage.Payload));
  47. }).UseConnectedHandler((e) =>
  48. {
  49. //MessageLog.GetInstance.Show($"连接成功");
  50. });
  51. try
  52. {
  53. await client.ConnectAsync(options);
  54. }
  55. catch (Exception ex)
  56. {
  57. MessageLog.GetInstance.ShowEx(ex.Message);
  58. MessageLog.GetInstance.Show("mqtt连接失败!重连执行中");
  59. }
  60. if (client.IsConnected)
  61. {
  62. MessageLog.GetInstance.Show("MQTT连接成功!");
  63. Connected?.Invoke();
  64. }
  65. }
  66. private void Reconnect()
  67. {
  68. UseDisconnectedAction = new Action(() =>
  69. {
  70. MessageLog.GetInstance.ShowEx("MQTT 断开连接");
  71. Thread.Sleep(2000);
  72. while (!UniversalHelper.GetInstance().GetNetworkState())
  73. {
  74. Thread.Sleep(2000);
  75. }
  76. bool ErrorFlag = false;
  77. while (!client.IsConnected)
  78. {
  79. try
  80. {
  81. MessageLog.GetInstance.Show($"重连中");
  82. client.ConnectAsync(options).Wait();
  83. }
  84. catch (Exception ex)
  85. {
  86. if (!ErrorFlag)
  87. {
  88. MessageLog.GetInstance.ShowEx(ex.ToString());
  89. ErrorFlag = true;
  90. }
  91. }
  92. Thread.Sleep(3000);
  93. }
  94. if (client.IsConnected)
  95. {
  96. MessageLog.GetInstance.Show("MQTT重连成功!");
  97. LostConnect?.Invoke();
  98. }
  99. UseDisconnectedAction = null;
  100. });
  101. }
  102. /// <summary>
  103. /// Mqtt 订阅
  104. /// </summary>
  105. /// <param name="topic">需要订阅的主题</param>
  106. public async void MqttSubscriptionAsync(string topic)
  107. {
  108. if (client != null && client.IsConnected)
  109. {
  110. try
  111. {
  112. var result = await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).WithExactlyOnceQoS().Build());
  113. }
  114. catch { }
  115. }
  116. }
  117. /// <summary>
  118. /// Mqtt 订阅
  119. /// </summary>
  120. /// <param name="topic">需要订阅的主题</param>
  121. public async void Subscrib(params string[] topic)
  122. {
  123. if (client != null && client.IsConnected)
  124. {
  125. try
  126. {
  127. foreach (var item in topic)
  128. {
  129. var result = await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(item).WithExactlyOnceQoS().Build());
  130. }
  131. }
  132. catch { }
  133. }
  134. }
  135. /// <summary>
  136. /// Mqtt 发布
  137. /// </summary>
  138. /// <param name="topic">需要发布的主题</param>
  139. /// <param name="content">需要发布的内容</param>
  140. public async void Publish(string topic, string content)
  141. {
  142. ;
  143. if (client != null && client.IsConnected)
  144. {
  145. push = true;
  146. var msg = new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(content).WithExactlyOnceQoS().Build();
  147. try
  148. {
  149. var result = await client.PublishAsync(msg);
  150. }
  151. catch { }
  152. }
  153. if(!client.IsConnected)
  154. {
  155. if(push)
  156. {
  157. MessageLog.GetInstance.ShowEx("MQTT未连接");
  158. push = false;
  159. }
  160. }
  161. }
  162. //public Action<string> MessageRecive { get; set; }
  163. //public Action Connected { get; set; }
  164. //public Action LostConnect { get; set; }
  165. //public bool IsConnected { get; set; }
  166. //private IMqttClient client;
  167. //public void Connect(string userName, string Password, string ip, int port, string clientId)
  168. //{
  169. // IConfigurationBuilder configurationBuilder = new ConfigurationBuilder();
  170. // configurationBuilder.AddMqttClientHostedService(p =>
  171. // {
  172. // p.Server = ip;
  173. // p.Port = port;
  174. // //p.UserName = "rafiul";
  175. // //p.Password = "12345678";
  176. // p.UserName = userName;
  177. // p.Password = Password;
  178. // p.mqttClientConnectedHandlerDelegate = new MQTTnet.Client.Connecting.MqttClientConnectedHandlerDelegate(e =>
  179. // {
  180. // IsConnected = true;
  181. // Connected?.Invoke();
  182. // });
  183. // //p.mqttClientDisconnectedHandlerDelegate = new MQTTnet.Client.Disconnecting.MqttClientDisconnectedHandlerDelegate(e =>
  184. // //{
  185. // // IsConnected = false;
  186. // // LostConnect?.Invoke();
  187. // //});
  188. // p.ConnectedResult += (s, e) =>
  189. // {
  190. // client = e;
  191. // };
  192. // p.MqttApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e =>
  193. // {
  194. // MessageRecive?.Invoke(Encoding.Default.GetString(e.ApplicationMessage.Payload));
  195. // });
  196. // });
  197. //}
  198. public void CloseConnect()
  199. {
  200. client.Dispose();
  201. }
  202. //public async void Publish(string topic, string content)
  203. //{
  204. // if (client.IsConnected)
  205. // await client.PublishAsync(topic, content);
  206. //}
  207. //public async void Subscrib(params string[] topics)
  208. //{
  209. // foreach (var topic in topics)
  210. // {
  211. // await client.SubscribeAsync(new MqttTopicFilter() { Topic = topic, QualityOfServiceLevel = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce });
  212. // }
  213. //}
  214. }
  215. }