Non puoi selezionare più di 25 argomenti Gli argomenti devono iniziare con una lettera o un numero, possono includere trattini ('-') e possono essere lunghi fino a 35 caratteri.
 
 
 
 

392 righe
14 KiB

  1. using System;
  2. using System.Collections.Concurrent;
  3. using System.Collections.Generic;
  4. using System.Linq;
  5. using System.Threading;
  6. using System.Threading.Tasks;
  7. using MQTTnet.Client;
  8. using MQTTnet.Diagnostics;
  9. using MQTTnet.Exceptions;
  10. using MQTTnet.Internal;
  11. using MQTTnet.Protocol;
  12. namespace MQTTnet.Extensions.ManagedClient
  13. {
  14. public class ManagedMqttClient : IManagedMqttClient
  15. {
  16. private readonly BlockingCollection<ManagedMqttApplicationMessage> _messageQueue = new BlockingCollection<ManagedMqttApplicationMessage>();
  17. private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
  18. private readonly AsyncLock _subscriptionsLock = new AsyncLock();
  19. private readonly List<string> _unsubscriptions = new List<string>();
  20. private readonly IMqttClient _mqttClient;
  21. private readonly IMqttNetChildLogger _logger;
  22. private CancellationTokenSource _connectionCancellationToken;
  23. private CancellationTokenSource _publishingCancellationToken;
  24. private ManagedMqttClientStorageManager _storageManager;
  25. private IManagedMqttClientOptions _options;
  26. private bool _subscriptionsNotPushed;
  27. public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
  28. {
  29. if (logger == null) throw new ArgumentNullException(nameof(logger));
  30. _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
  31. _mqttClient.Connected += OnConnected;
  32. _mqttClient.Disconnected += OnDisconnected;
  33. _mqttClient.ApplicationMessageReceived += OnApplicationMessageReceived;
  34. _logger = logger.CreateChildLogger(nameof(ManagedMqttClient));
  35. }
  36. public bool IsConnected => _mqttClient.IsConnected;
  37. public bool IsStarted => _connectionCancellationToken != null;
  38. public event EventHandler<MqttClientConnectedEventArgs> Connected;
  39. public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
  40. public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
  41. public event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
  42. public event EventHandler SynchronizingSubscriptionsFailed;
  43. public async Task StartAsync(IManagedMqttClientOptions options)
  44. {
  45. if (options == null) throw new ArgumentNullException(nameof(options));
  46. if (options.ClientOptions == null) throw new ArgumentException("The client options are not set.", nameof(options));
  47. if (!options.ClientOptions.CleanSession)
  48. {
  49. throw new NotSupportedException("The managed client does not support existing sessions.");
  50. }
  51. if (_connectionCancellationToken != null) throw new InvalidOperationException("The managed client is already started.");
  52. _options = options;
  53. if (_options.Storage != null)
  54. {
  55. _storageManager = new ManagedMqttClientStorageManager(_options.Storage);
  56. var messages = await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
  57. foreach (var message in messages)
  58. {
  59. _messageQueue.Add(message);
  60. }
  61. }
  62. _connectionCancellationToken = new CancellationTokenSource();
  63. #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  64. Task.Run(() => MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token);
  65. #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  66. _logger.Info("Started");
  67. }
  68. public Task StopAsync()
  69. {
  70. StopPublishing();
  71. StopMaintainingConnection();
  72. while (_messageQueue.Any())
  73. {
  74. _messageQueue.Take();
  75. }
  76. return Task.FromResult(0);
  77. }
  78. public Task PublishAsync(MqttApplicationMessage applicationMessage)
  79. {
  80. if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
  81. return PublishAsync(new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(applicationMessage).Build());
  82. }
  83. public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage)
  84. {
  85. if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
  86. if (_storageManager != null)
  87. {
  88. await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
  89. }
  90. _messageQueue.Add(applicationMessage);
  91. }
  92. public async Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
  93. {
  94. if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
  95. using (await _subscriptionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
  96. {
  97. foreach (var topicFilter in topicFilters)
  98. {
  99. _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
  100. _subscriptionsNotPushed = true;
  101. }
  102. }
  103. }
  104. public async Task UnsubscribeAsync(IEnumerable<string> topics)
  105. {
  106. using (await _subscriptionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
  107. {
  108. foreach (var topic in topics)
  109. {
  110. if (_subscriptions.Remove(topic))
  111. {
  112. _unsubscriptions.Add(topic);
  113. _subscriptionsNotPushed = true;
  114. }
  115. }
  116. }
  117. }
  118. public void Dispose()
  119. {
  120. _messageQueue?.Dispose();
  121. _subscriptionsLock?.Dispose();
  122. _connectionCancellationToken?.Dispose();
  123. _publishingCancellationToken?.Dispose();
  124. }
  125. private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
  126. {
  127. try
  128. {
  129. while (!cancellationToken.IsCancellationRequested)
  130. {
  131. await TryMaintainConnectionAsync(cancellationToken).ConfigureAwait(false);
  132. }
  133. }
  134. catch (OperationCanceledException)
  135. {
  136. }
  137. catch (Exception exception)
  138. {
  139. _logger.Error(exception, "Unhandled exception while maintaining connection.");
  140. }
  141. finally
  142. {
  143. await _mqttClient.DisconnectAsync().ConfigureAwait(false);
  144. _logger.Info("Stopped");
  145. }
  146. }
  147. private async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
  148. {
  149. try
  150. {
  151. var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false);
  152. if (connectionState == ReconnectionResult.NotConnected)
  153. {
  154. StopPublishing();
  155. await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
  156. return;
  157. }
  158. if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
  159. {
  160. await SynchronizeSubscriptionsAsync().ConfigureAwait(false);
  161. StartPublishing();
  162. return;
  163. }
  164. if (connectionState == ReconnectionResult.StillConnected)
  165. {
  166. await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
  167. }
  168. }
  169. catch (OperationCanceledException)
  170. {
  171. }
  172. catch (MqttCommunicationException exception)
  173. {
  174. _logger.Warning(exception, "Communication exception while maintaining connection.");
  175. }
  176. catch (Exception exception)
  177. {
  178. _logger.Error(exception, "Unhandled exception while maintaining connection.");
  179. }
  180. }
  181. private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
  182. {
  183. try
  184. {
  185. while (!cancellationToken.IsCancellationRequested)
  186. {
  187. var message = _messageQueue.Take(cancellationToken);
  188. if (message == null)
  189. {
  190. continue;
  191. }
  192. if (cancellationToken.IsCancellationRequested)
  193. {
  194. continue;
  195. }
  196. await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
  197. }
  198. }
  199. catch (OperationCanceledException)
  200. {
  201. }
  202. catch (Exception exception)
  203. {
  204. _logger.Error(exception, "Unhandled exception while publishing queued application messages.");
  205. }
  206. finally
  207. {
  208. _logger.Verbose("Stopped publishing messages.");
  209. }
  210. }
  211. private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
  212. {
  213. Exception transmitException = null;
  214. try
  215. {
  216. await _mqttClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
  217. if (_storageManager != null)
  218. {
  219. await _storageManager.RemoveAsync(message).ConfigureAwait(false);
  220. }
  221. }
  222. catch (MqttCommunicationException exception)
  223. {
  224. transmitException = exception;
  225. _logger.Warning(exception, $"Publishing application ({message.Id}) message failed.");
  226. if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
  227. {
  228. _messageQueue.Add(message);
  229. }
  230. }
  231. catch (Exception exception)
  232. {
  233. transmitException = exception;
  234. _logger.Error(exception, $"Unhandled exception while publishing application message ({message.Id}).");
  235. }
  236. finally
  237. {
  238. ApplicationMessageProcessed?.Invoke(this, new ApplicationMessageProcessedEventArgs(message, transmitException));
  239. }
  240. }
  241. private async Task SynchronizeSubscriptionsAsync()
  242. {
  243. _logger.Info(nameof(ManagedMqttClient), "Synchronizing subscriptions");
  244. List<TopicFilter> subscriptions;
  245. List<string> unsubscriptions;
  246. using (await _subscriptionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
  247. {
  248. subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList();
  249. unsubscriptions = new List<string>(_unsubscriptions);
  250. _unsubscriptions.Clear();
  251. _subscriptionsNotPushed = false;
  252. }
  253. if (!subscriptions.Any() && !unsubscriptions.Any())
  254. {
  255. return;
  256. }
  257. try
  258. {
  259. if (subscriptions.Any())
  260. {
  261. await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false);
  262. }
  263. if (unsubscriptions.Any())
  264. {
  265. await _mqttClient.UnsubscribeAsync(unsubscriptions).ConfigureAwait(false);
  266. }
  267. }
  268. catch (Exception exception)
  269. {
  270. _logger.Warning(exception, "Synchronizing subscriptions failed.");
  271. _subscriptionsNotPushed = true;
  272. SynchronizingSubscriptionsFailed?.Invoke(this, EventArgs.Empty);
  273. }
  274. }
  275. private async Task<ReconnectionResult> ReconnectIfRequiredAsync()
  276. {
  277. if (_mqttClient.IsConnected)
  278. {
  279. return ReconnectionResult.StillConnected;
  280. }
  281. try
  282. {
  283. await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false);
  284. return ReconnectionResult.Reconnected;
  285. }
  286. catch (Exception)
  287. {
  288. return ReconnectionResult.NotConnected;
  289. }
  290. }
  291. private void OnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs eventArgs)
  292. {
  293. ApplicationMessageReceived?.Invoke(this, eventArgs);
  294. }
  295. private void OnDisconnected(object sender, MqttClientDisconnectedEventArgs eventArgs)
  296. {
  297. Disconnected?.Invoke(this, eventArgs);
  298. }
  299. private void OnConnected(object sender, MqttClientConnectedEventArgs eventArgs)
  300. {
  301. Connected?.Invoke(this, eventArgs);
  302. }
  303. private void StartPublishing()
  304. {
  305. if (_publishingCancellationToken != null)
  306. {
  307. StopPublishing();
  308. }
  309. var cts = new CancellationTokenSource();
  310. _publishingCancellationToken = cts;
  311. #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  312. Task.Run(() => PublishQueuedMessagesAsync(cts.Token), cts.Token);
  313. #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  314. }
  315. private void StopPublishing()
  316. {
  317. _publishingCancellationToken?.Cancel(false);
  318. _publishingCancellationToken?.Dispose();
  319. _publishingCancellationToken = null;
  320. }
  321. private void StopMaintainingConnection()
  322. {
  323. _connectionCancellationToken?.Cancel(false);
  324. _connectionCancellationToken?.Dispose();
  325. _connectionCancellationToken = null;
  326. }
  327. }
  328. }