您最多选择25个主题 主题必须以字母或数字开头,可以包含连字符 (-),并且长度不得超过35个字符

ManagedMqttClient.cs 20 KiB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq;
  4. using System.Threading;
  5. using System.Threading.Tasks;
  6. using MQTTnet.Client;
  7. using MQTTnet.Client.Connecting;
  8. using MQTTnet.Client.Disconnecting;
  9. using MQTTnet.Client.Publishing;
  10. using MQTTnet.Client.Receiving;
  11. using MQTTnet.Diagnostics;
  12. using MQTTnet.Exceptions;
  13. using MQTTnet.Internal;
  14. using MQTTnet.Protocol;
  15. using MQTTnet.Server;
  16. namespace MQTTnet.Extensions.ManagedClient
  17. {
  18. public class ManagedMqttClient : IManagedMqttClient
  19. {
  20. private readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
  21. private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
  22. private readonly HashSet<string> _unsubscriptions = new HashSet<string>();
  23. private readonly IMqttClient _mqttClient;
  24. private readonly IMqttNetChildLogger _logger;
  25. private CancellationTokenSource _connectionCancellationToken;
  26. private CancellationTokenSource _publishingCancellationToken;
  27. private Task _maintainConnectionTask;
  28. private ManagedMqttClientStorageManager _storageManager;
  29. private bool _disposed = false;
  30. private bool _subscriptionsNotPushed;
  31. public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
  32. {
  33. _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
  34. if (logger == null) throw new ArgumentNullException(nameof(logger));
  35. _logger = logger.CreateChildLogger(nameof(ManagedMqttClient));
  36. }
  37. public bool IsConnected => _mqttClient.IsConnected;
  38. public bool IsStarted => _connectionCancellationToken != null;
  39. public int PendingApplicationMessagesCount => _messageQueue.Count;
  40. public IManagedMqttClientOptions Options { get; private set; }
  41. public IMqttClientConnectedHandler ConnectedHandler
  42. {
  43. get => _mqttClient.ConnectedHandler;
  44. set => _mqttClient.ConnectedHandler = value;
  45. }
  46. public IMqttClientDisconnectedHandler DisconnectedHandler
  47. {
  48. get => _mqttClient.DisconnectedHandler;
  49. set => _mqttClient.DisconnectedHandler = value;
  50. }
  51. public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
  52. {
  53. get => _mqttClient.ApplicationMessageReceivedHandler;
  54. set => _mqttClient.ApplicationMessageReceivedHandler = value;
  55. }
  56. public IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; }
  57. public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; }
  58. public IConnectingFailedHandler ConnectingFailedHandler { get; set; }
  59. public ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; }
  60. public async Task StartAsync(IManagedMqttClientOptions options)
  61. {
  62. ThrowIfDisposed();
  63. if (options == null) throw new ArgumentNullException(nameof(options));
  64. if (options.ClientOptions == null) throw new ArgumentException("The client options are not set.", nameof(options));
  65. if (!options.ClientOptions.CleanSession)
  66. {
  67. throw new NotSupportedException("The managed client does not support existing sessions.");
  68. }
  69. if (!_maintainConnectionTask?.IsCompleted ?? false) throw new InvalidOperationException("The managed client is already started.");
  70. Options = options;
  71. if (Options.Storage != null)
  72. {
  73. _storageManager = new ManagedMqttClientStorageManager(Options.Storage);
  74. var messages = await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
  75. foreach (var message in messages)
  76. {
  77. _messageQueue.Enqueue(message);
  78. }
  79. }
  80. _connectionCancellationToken = new CancellationTokenSource();
  81. #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  82. _maintainConnectionTask = Task.Run(() => MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token);
  83. #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  84. _logger.Info("Started");
  85. }
  86. public async Task StopAsync()
  87. {
  88. ThrowIfDisposed();
  89. StopPublishing();
  90. StopMaintainingConnection();
  91. _messageQueue.Clear();
  92. if (_maintainConnectionTask != null)
  93. {
  94. await Task.WhenAny(_maintainConnectionTask);
  95. _maintainConnectionTask = null;
  96. }
  97. }
  98. public async Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
  99. {
  100. ThrowIfDisposed();
  101. if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
  102. await PublishAsync(new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(applicationMessage).Build()).ConfigureAwait(false);
  103. return new MqttClientPublishResult();
  104. }
  105. public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage)
  106. {
  107. ThrowIfDisposed();
  108. if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
  109. MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage.Topic);
  110. ManagedMqttApplicationMessage removedMessage = null;
  111. ApplicationMessageSkippedEventArgs applicationMessageSkippedEventArgs = null;
  112. try
  113. {
  114. lock (_messageQueue)
  115. {
  116. if (_messageQueue.Count >= Options.MaxPendingMessages)
  117. {
  118. if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
  119. {
  120. _logger.Verbose("Skipping publish of new application message because internal queue is full.");
  121. applicationMessageSkippedEventArgs = new ApplicationMessageSkippedEventArgs(applicationMessage);
  122. return;
  123. }
  124. if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
  125. {
  126. removedMessage = _messageQueue.RemoveFirst();
  127. _logger.Verbose("Removed oldest application message from internal queue because it is full.");
  128. applicationMessageSkippedEventArgs = new ApplicationMessageSkippedEventArgs(removedMessage);
  129. }
  130. }
  131. _messageQueue.Enqueue(applicationMessage);
  132. }
  133. }
  134. finally
  135. {
  136. if (applicationMessageSkippedEventArgs != null)
  137. {
  138. var applicationMessageSkippedHandler = ApplicationMessageSkippedHandler;
  139. if (applicationMessageSkippedHandler != null)
  140. {
  141. await applicationMessageSkippedHandler.HandleApplicationMessageSkippedAsync(applicationMessageSkippedEventArgs).ConfigureAwait(false);
  142. }
  143. }
  144. }
  145. if (_storageManager != null)
  146. {
  147. if (removedMessage != null)
  148. {
  149. await _storageManager.RemoveAsync(removedMessage).ConfigureAwait(false);
  150. }
  151. await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
  152. }
  153. }
  154. public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
  155. {
  156. ThrowIfDisposed();
  157. if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
  158. lock (_subscriptions)
  159. {
  160. foreach (var topicFilter in topicFilters)
  161. {
  162. _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
  163. _subscriptionsNotPushed = true;
  164. }
  165. }
  166. return Task.FromResult(0);
  167. }
  168. public Task UnsubscribeAsync(IEnumerable<string> topics)
  169. {
  170. ThrowIfDisposed();
  171. if (topics == null) throw new ArgumentNullException(nameof(topics));
  172. lock (_subscriptions)
  173. {
  174. foreach (var topic in topics)
  175. {
  176. if (_subscriptions.Remove(topic))
  177. {
  178. _unsubscriptions.Add(topic);
  179. _subscriptionsNotPushed = true;
  180. }
  181. }
  182. }
  183. return Task.FromResult(0);
  184. }
  185. public void Dispose()
  186. {
  187. if (_disposed)
  188. {
  189. return;
  190. }
  191. _disposed = true;
  192. StopPublishing();
  193. StopMaintainingConnection();
  194. if (_maintainConnectionTask != null)
  195. {
  196. Task.WaitAny(_maintainConnectionTask);
  197. _maintainConnectionTask = null;
  198. }
  199. _mqttClient.Dispose();
  200. }
  201. private void ThrowIfDisposed()
  202. {
  203. if (_disposed)
  204. {
  205. throw new ObjectDisposedException(nameof(ManagedMqttClient));
  206. }
  207. }
  208. private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
  209. {
  210. try
  211. {
  212. while (!cancellationToken.IsCancellationRequested)
  213. {
  214. await TryMaintainConnectionAsync(cancellationToken).ConfigureAwait(false);
  215. }
  216. }
  217. catch (OperationCanceledException)
  218. {
  219. }
  220. catch (Exception exception)
  221. {
  222. _logger.Error(exception, "Unhandled exception while maintaining connection.");
  223. }
  224. finally
  225. {
  226. if (!_disposed)
  227. {
  228. try
  229. {
  230. await _mqttClient.DisconnectAsync().ConfigureAwait(false);
  231. }
  232. catch (Exception exception)
  233. {
  234. _logger.Error(exception, "Error while disconnecting.");
  235. }
  236. _logger.Info("Stopped");
  237. }
  238. }
  239. }
  240. private async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
  241. {
  242. try
  243. {
  244. var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false);
  245. if (connectionState == ReconnectionResult.NotConnected)
  246. {
  247. StopPublishing();
  248. await Task.Delay(Options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
  249. return;
  250. }
  251. if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
  252. {
  253. await SynchronizeSubscriptionsAsync().ConfigureAwait(false);
  254. StartPublishing();
  255. return;
  256. }
  257. if (connectionState == ReconnectionResult.StillConnected)
  258. {
  259. await Task.Delay(Options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false);
  260. }
  261. }
  262. catch (OperationCanceledException)
  263. {
  264. }
  265. catch (MqttCommunicationException exception)
  266. {
  267. _logger.Warning(exception, "Communication exception while maintaining connection.");
  268. }
  269. catch (Exception exception)
  270. {
  271. _logger.Error(exception, "Unhandled exception while maintaining connection.");
  272. }
  273. }
  274. private void PublishQueuedMessages(CancellationToken cancellationToken)
  275. {
  276. try
  277. {
  278. while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected)
  279. {
  280. //Peek at the message without dequeueing in order to prevent the
  281. //possibility of the queue growing beyond the configured cap.
  282. //Previously, messages could be re-enqueued if there was an
  283. //exception, and this re-enqueueing did not honor the cap.
  284. //Furthermore, because re-enqueueing would shuffle the order
  285. //of the messages, the DropOldestQueuedMessage strategy would
  286. //be unable to know which message is actually the oldest and would
  287. //instead drop the first item in the queue.
  288. var message = _messageQueue.PeekAndWait();
  289. if (message == null)
  290. {
  291. continue;
  292. }
  293. cancellationToken.ThrowIfCancellationRequested();
  294. TryPublishQueuedMessage(message);
  295. }
  296. }
  297. catch (OperationCanceledException)
  298. {
  299. }
  300. catch (Exception exception)
  301. {
  302. _logger.Error(exception, "Unhandled exception while publishing queued application messages.");
  303. }
  304. finally
  305. {
  306. _logger.Verbose("Stopped publishing messages.");
  307. }
  308. }
  309. private void TryPublishQueuedMessage(ManagedMqttApplicationMessage message)
  310. {
  311. Exception transmitException = null;
  312. try
  313. {
  314. _mqttClient.PublishAsync(message.ApplicationMessage).GetAwaiter().GetResult();
  315. lock (_messageQueue) //lock to avoid conflict with this.PublishAsync
  316. {
  317. //While publishing this message, this.PublishAsync could have booted this
  318. //message off the queue to make room for another (when using a cap
  319. //with the DropOldestQueuedMessage strategy). If the first item
  320. //in the queue is equal to this message, then it's safe to remove
  321. //it from the queue. If not, that means this.PublishAsync has already
  322. //removed it, in which case we don't want to do anything.
  323. _messageQueue.RemoveFirst(i => i.Id.Equals(message.Id));
  324. }
  325. _storageManager?.RemoveAsync(message).GetAwaiter().GetResult();
  326. }
  327. catch (MqttCommunicationException exception)
  328. {
  329. transmitException = exception;
  330. _logger.Warning(exception, $"Publishing application ({message.Id}) message failed.");
  331. if (message.ApplicationMessage.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
  332. {
  333. //If QoS 0, we don't want this message to stay on the queue.
  334. //If QoS 1 or 2, it's possible that, when using a cap, this message
  335. //has been booted off the queue by this.PublishAsync, in which case this
  336. //thread will not continue to try to publish it. While this does
  337. //contradict the expected behavior of QoS 1 and 2, that's also true
  338. //for the usage of a message queue cap, so it's still consistent
  339. //with prior behavior in that way.
  340. lock (_messageQueue) //lock to avoid conflict with this.PublishAsync
  341. {
  342. _messageQueue.RemoveFirst(i => i.Id.Equals(message.Id));
  343. }
  344. }
  345. }
  346. catch (Exception exception)
  347. {
  348. transmitException = exception;
  349. _logger.Error(exception, $"Unhandled exception while publishing application message ({message.Id}).");
  350. }
  351. finally
  352. {
  353. ApplicationMessageProcessedHandler?.HandleApplicationMessageProcessedAsync(new ApplicationMessageProcessedEventArgs(message, transmitException)).GetAwaiter().GetResult();
  354. }
  355. }
  356. private async Task SynchronizeSubscriptionsAsync()
  357. {
  358. _logger.Info("Synchronizing subscriptions");
  359. List<TopicFilter> subscriptions;
  360. HashSet<string> unsubscriptions;
  361. lock (_subscriptions)
  362. {
  363. subscriptions = _subscriptions.Select(i => new TopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }).ToList();
  364. unsubscriptions = new HashSet<string>(_unsubscriptions);
  365. _unsubscriptions.Clear();
  366. _subscriptionsNotPushed = false;
  367. }
  368. if (!subscriptions.Any() && !unsubscriptions.Any())
  369. {
  370. return;
  371. }
  372. try
  373. {
  374. if (unsubscriptions.Any())
  375. {
  376. await _mqttClient.UnsubscribeAsync(unsubscriptions.ToArray()).ConfigureAwait(false);
  377. }
  378. if (subscriptions.Any())
  379. {
  380. await _mqttClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false);
  381. }
  382. }
  383. catch (Exception exception)
  384. {
  385. _logger.Warning(exception, "Synchronizing subscriptions failed.");
  386. _subscriptionsNotPushed = true;
  387. var synchronizingSubscriptionsFailedHandler = SynchronizingSubscriptionsFailedHandler;
  388. if (SynchronizingSubscriptionsFailedHandler != null)
  389. {
  390. await synchronizingSubscriptionsFailedHandler.HandleSynchronizingSubscriptionsFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false);
  391. }
  392. }
  393. }
  394. private async Task<ReconnectionResult> ReconnectIfRequiredAsync()
  395. {
  396. if (_mqttClient.IsConnected)
  397. {
  398. return ReconnectionResult.StillConnected;
  399. }
  400. try
  401. {
  402. await _mqttClient.ConnectAsync(Options.ClientOptions).ConfigureAwait(false);
  403. return ReconnectionResult.Reconnected;
  404. }
  405. catch (Exception exception)
  406. {
  407. var connectingFailedHandler = ConnectingFailedHandler;
  408. if (connectingFailedHandler != null)
  409. {
  410. await connectingFailedHandler.HandleConnectingFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false);
  411. }
  412. return ReconnectionResult.NotConnected;
  413. }
  414. }
  415. private void StartPublishing()
  416. {
  417. if (_publishingCancellationToken != null)
  418. {
  419. StopPublishing();
  420. }
  421. var cts = new CancellationTokenSource();
  422. _publishingCancellationToken = cts;
  423. Task.Factory.StartNew(() => PublishQueuedMessages(cts.Token), cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
  424. }
  425. private void StopPublishing()
  426. {
  427. _publishingCancellationToken?.Cancel(false);
  428. _publishingCancellationToken?.Dispose();
  429. _publishingCancellationToken = null;
  430. }
  431. private void StopMaintainingConnection()
  432. {
  433. _connectionCancellationToken?.Cancel(false);
  434. _connectionCancellationToken?.Dispose();
  435. _connectionCancellationToken = null;
  436. }
  437. }
  438. }