選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。
 
 
 
 

468 行
19 KiB

  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq;
  4. using System.Threading;
  5. using System.Threading.Tasks;
  6. using MQTTnet.Core.Adapter;
  7. using MQTTnet.Core.Diagnostics;
  8. using MQTTnet.Core.Exceptions;
  9. using MQTTnet.Core.Internal;
  10. using MQTTnet.Core.Packets;
  11. using MQTTnet.Core.Protocol;
  12. namespace MQTTnet.Core.Client
  13. {
  14. public class MqttClient : IMqttClient
  15. {
  16. private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
  17. private readonly MqttPacketDispatcher _packetDispatcher;
  18. private readonly IMqttCommunicationAdapterFactory _communicationChannelFactory;
  19. private readonly MqttNetTrace _trace;
  20. private IMqttClientOptions _options;
  21. private bool _isReceivingPackets;
  22. private int _latestPacketIdentifier;
  23. private CancellationTokenSource _cancellationTokenSource;
  24. private IMqttCommunicationAdapter _adapter;
  25. public MqttClient(IMqttCommunicationAdapterFactory communicationChannelFactory, MqttNetTrace trace)
  26. {
  27. _trace = trace ?? throw new ArgumentNullException(nameof(trace));
  28. _communicationChannelFactory = communicationChannelFactory ?? throw new ArgumentNullException(nameof(communicationChannelFactory));
  29. _packetDispatcher = new MqttPacketDispatcher(trace);
  30. }
  31. public event EventHandler<MqttClientConnectedEventArgs> Connected;
  32. public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
  33. public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
  34. public bool IsConnected { get; private set; }
  35. public async Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options)
  36. {
  37. if (options == null) throw new ArgumentNullException(nameof(options));
  38. ThrowIfConnected("It is not allowed to connect with a server after the connection is established.");
  39. try
  40. {
  41. _options = options;
  42. _cancellationTokenSource = new CancellationTokenSource();
  43. _latestPacketIdentifier = 0;
  44. _packetDispatcher.Reset();
  45. _adapter = _communicationChannelFactory.CreateMqttCommunicationAdapter(options);
  46. _trace.Verbose(nameof(MqttClient), "Trying to connect with server.");
  47. await _adapter.ConnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
  48. _trace.Verbose(nameof(MqttClient), "Connection with server established.");
  49. await SetupIncomingPacketProcessingAsync();
  50. var connectResponse = await AuthenticateAsync(options.WillMessage);
  51. _trace.Verbose(nameof(MqttClient), "MQTT connection with server established.");
  52. if (_options.KeepAlivePeriod != TimeSpan.Zero)
  53. {
  54. StartSendKeepAliveMessages(_cancellationTokenSource.Token);
  55. }
  56. IsConnected = true;
  57. Connected?.Invoke(this, new MqttClientConnectedEventArgs(connectResponse.IsSessionPresent));
  58. return new MqttClientConnectResult(connectResponse.IsSessionPresent);
  59. }
  60. catch (Exception)
  61. {
  62. await DisconnectInternalAsync().ConfigureAwait(false);
  63. throw;
  64. }
  65. }
  66. public async Task DisconnectAsync()
  67. {
  68. if (!IsConnected)
  69. {
  70. return;
  71. }
  72. try
  73. {
  74. await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false);
  75. }
  76. finally
  77. {
  78. await DisconnectInternalAsync().ConfigureAwait(false);
  79. }
  80. }
  81. public async Task<IList<MqttSubscribeResult>> SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
  82. {
  83. if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
  84. ThrowIfNotConnected();
  85. var subscribePacket = new MqttSubscribePacket
  86. {
  87. PacketIdentifier = GetNewPacketIdentifier(),
  88. TopicFilters = topicFilters.ToList()
  89. };
  90. var response = await SendAndReceiveAsync<MqttSubAckPacket>(subscribePacket).ConfigureAwait(false);
  91. if (response.SubscribeReturnCodes.Count != subscribePacket.TopicFilters.Count)
  92. {
  93. throw new MqttProtocolViolationException("The return codes are not matching the topic filters [MQTT-3.9.3-1].");
  94. }
  95. return subscribePacket.TopicFilters.Select((t, i) => new MqttSubscribeResult(t, response.SubscribeReturnCodes[i])).ToList();
  96. }
  97. public async Task UnsubscribeAsync(IEnumerable<string> topicFilters)
  98. {
  99. if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
  100. ThrowIfNotConnected();
  101. var unsubscribePacket = new MqttUnsubscribePacket
  102. {
  103. PacketIdentifier = GetNewPacketIdentifier(),
  104. TopicFilters = topicFilters.ToList()
  105. };
  106. await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket);
  107. }
  108. public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
  109. {
  110. ThrowIfNotConnected();
  111. var publishPackets = applicationMessages.Select(m => m.ToPublishPacket());
  112. foreach (var qosGroup in publishPackets.GroupBy(p => p.QualityOfServiceLevel))
  113. {
  114. var qosPackets = qosGroup.ToArray();
  115. switch (qosGroup.Key)
  116. {
  117. case MqttQualityOfServiceLevel.AtMostOnce:
  118. {
  119. // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
  120. await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, qosPackets);
  121. break;
  122. }
  123. case MqttQualityOfServiceLevel.AtLeastOnce:
  124. {
  125. foreach (var publishPacket in qosPackets)
  126. {
  127. publishPacket.PacketIdentifier = GetNewPacketIdentifier();
  128. await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket);
  129. }
  130. break;
  131. }
  132. case MqttQualityOfServiceLevel.ExactlyOnce:
  133. {
  134. foreach (var publishPacket in qosPackets)
  135. {
  136. publishPacket.PacketIdentifier = GetNewPacketIdentifier();
  137. var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
  138. await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
  139. }
  140. break;
  141. }
  142. default:
  143. {
  144. throw new InvalidOperationException();
  145. }
  146. }
  147. }
  148. }
  149. private async Task<MqttConnAckPacket> AuthenticateAsync(MqttApplicationMessage willApplicationMessage)
  150. {
  151. var connectPacket = new MqttConnectPacket
  152. {
  153. ClientId = _options.ClientId,
  154. Username = _options.UserName,
  155. Password = _options.Password,
  156. CleanSession = _options.CleanSession,
  157. KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds,
  158. WillMessage = willApplicationMessage
  159. };
  160. var response = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket).ConfigureAwait(false);
  161. if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
  162. {
  163. throw new MqttConnectingFailedException(response.ConnectReturnCode);
  164. }
  165. return response;
  166. }
  167. private async Task SetupIncomingPacketProcessingAsync()
  168. {
  169. _isReceivingPackets = false;
  170. #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  171. Task.Factory.StartNew(
  172. () => ReceivePackets(_cancellationTokenSource.Token),
  173. _cancellationTokenSource.Token,
  174. TaskCreationOptions.LongRunning,
  175. TaskScheduler.Default).ConfigureAwait(false);
  176. #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  177. while (!_isReceivingPackets && _cancellationTokenSource != null && !_cancellationTokenSource.IsCancellationRequested)
  178. {
  179. await Task.Delay(TimeSpan.FromMilliseconds(100));
  180. }
  181. }
  182. private void ThrowIfNotConnected()
  183. {
  184. if (!IsConnected) throw new MqttCommunicationException("The client is not connected.");
  185. }
  186. private void ThrowIfConnected(string message)
  187. {
  188. if (IsConnected) throw new MqttProtocolViolationException(message);
  189. }
  190. private async Task DisconnectInternalAsync()
  191. {
  192. var clientWasConnected = IsConnected;
  193. IsConnected = false;
  194. var cts = _cancellationTokenSource;
  195. if (cts == null || cts.IsCancellationRequested)
  196. {
  197. return;
  198. }
  199. cts.Cancel(false);
  200. cts.Dispose();
  201. _cancellationTokenSource = null;
  202. try
  203. {
  204. await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
  205. _trace.Information(nameof(MqttClient), "Disconnected from adapter.");
  206. }
  207. catch (Exception exception)
  208. {
  209. _trace.Warning(nameof(MqttClient), exception, "Error while disconnecting from adapter.");
  210. }
  211. finally
  212. {
  213. Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected));
  214. }
  215. }
  216. private async Task ProcessReceivedPacketAsync(MqttBasePacket packet)
  217. {
  218. try
  219. {
  220. _trace.Information(nameof(MqttClient), "Received <<< {0}", packet);
  221. if (packet is MqttPingReqPacket)
  222. {
  223. await SendAsync(new MqttPingRespPacket());
  224. return;
  225. }
  226. if (packet is MqttDisconnectPacket)
  227. {
  228. await DisconnectAsync();
  229. return;
  230. }
  231. if (packet is MqttPublishPacket publishPacket)
  232. {
  233. await ProcessReceivedPublishPacket(publishPacket);
  234. return;
  235. }
  236. if (packet is MqttPubRelPacket pubRelPacket)
  237. {
  238. await ProcessReceivedPubRelPacket(pubRelPacket);
  239. return;
  240. }
  241. _packetDispatcher.Dispatch(packet);
  242. }
  243. catch (Exception exception)
  244. {
  245. _trace.Error(nameof(MqttClient), exception, "Unhandled exception while processing received packet.");
  246. }
  247. }
  248. private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
  249. {
  250. try
  251. {
  252. var applicationMessage = publishPacket.ToApplicationMessage();
  253. ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
  254. }
  255. catch (Exception exception)
  256. {
  257. _trace.Error(nameof(MqttClient), exception, "Unhandled exception while handling application message.");
  258. }
  259. }
  260. private async Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
  261. {
  262. if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
  263. {
  264. FireApplicationMessageReceivedEvent(publishPacket);
  265. return;
  266. }
  267. if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
  268. {
  269. FireApplicationMessageReceivedEvent(publishPacket);
  270. await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
  271. return;
  272. }
  273. if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
  274. {
  275. // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
  276. lock (_unacknowledgedPublishPackets)
  277. {
  278. _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
  279. }
  280. FireApplicationMessageReceivedEvent(publishPacket);
  281. await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
  282. return;
  283. }
  284. throw new MqttCommunicationException("Received a not supported QoS level.");
  285. }
  286. private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)
  287. {
  288. lock (_unacknowledgedPublishPackets)
  289. {
  290. _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
  291. }
  292. return SendAsync(pubRelPacket.CreateResponse<MqttPubCompPacket>());
  293. }
  294. private Task SendAsync(MqttBasePacket packet)
  295. {
  296. return _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, packet);
  297. }
  298. private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
  299. {
  300. var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout);
  301. await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false);
  302. return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
  303. }
  304. private ushort GetNewPacketIdentifier()
  305. {
  306. return (ushort)Interlocked.Increment(ref _latestPacketIdentifier);
  307. }
  308. private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
  309. {
  310. _trace.Information(nameof(MqttClient), "Start sending keep alive packets.");
  311. try
  312. {
  313. while (!cancellationToken.IsCancellationRequested)
  314. {
  315. await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false);
  316. if (cancellationToken.IsCancellationRequested)
  317. {
  318. return;
  319. }
  320. await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
  321. }
  322. }
  323. catch (OperationCanceledException)
  324. {
  325. }
  326. catch (MqttCommunicationException exception)
  327. {
  328. if (cancellationToken.IsCancellationRequested)
  329. {
  330. return;
  331. }
  332. _trace.Warning(nameof(MqttClient), exception, "MQTT communication exception while sending/receiving keep alive packets.");
  333. await DisconnectInternalAsync().ConfigureAwait(false);
  334. }
  335. catch (Exception exception)
  336. {
  337. _trace.Warning(nameof(MqttClient), exception, "Unhandled exception while sending/receiving keep alive packets.");
  338. await DisconnectInternalAsync().ConfigureAwait(false);
  339. }
  340. finally
  341. {
  342. _trace.Information(nameof(MqttClient), "Stopped sending keep alive packets.");
  343. }
  344. }
  345. private async Task ReceivePackets(CancellationToken cancellationToken)
  346. {
  347. _trace.Information(nameof(MqttClient), "Start receiving packets.");
  348. try
  349. {
  350. while (!cancellationToken.IsCancellationRequested)
  351. {
  352. _isReceivingPackets = true;
  353. var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
  354. if (cancellationToken.IsCancellationRequested)
  355. {
  356. return;
  357. }
  358. StartProcessReceivedPacket(packet, cancellationToken);
  359. }
  360. }
  361. catch (OperationCanceledException)
  362. {
  363. }
  364. catch (MqttCommunicationException exception)
  365. {
  366. if (cancellationToken.IsCancellationRequested)
  367. {
  368. return;
  369. }
  370. _trace.Warning(nameof(MqttClient), exception, "MQTT communication exception while receiving packets.");
  371. await DisconnectInternalAsync().ConfigureAwait(false);
  372. }
  373. catch (Exception exception)
  374. {
  375. _trace.Error(nameof(MqttClient), exception, "Unhandled exception while receiving packets.");
  376. await DisconnectInternalAsync().ConfigureAwait(false);
  377. }
  378. finally
  379. {
  380. _trace.Information(nameof(MqttClient), "Stopped receiving packets.");
  381. }
  382. }
  383. private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken)
  384. {
  385. #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  386. Task.Run(() => ProcessReceivedPacketAsync(packet), cancellationToken).ConfigureAwait(false);
  387. #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  388. }
  389. private void StartSendKeepAliveMessages(CancellationToken cancellationToken)
  390. {
  391. #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  392. Task.Factory.StartNew(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false);
  393. #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
  394. }
  395. }
  396. }