diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 664b09cd..4245077b 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -2,7 +2,7 @@ MQTTnet.AspNetCore - 2.7.4 + 2.7.5 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,13 +10,13 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - * Updated to MQTTnet 2.7.4. + * Updated to MQTTnet 2.7.5. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index 2f2fe9b7..8bca0b57 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -2,7 +2,7 @@ MQTTnet.Extensions.Rpc - 2.7.4 + 2.7.5 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,28 +10,28 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a extension library which allows executing synchronous device calls including a response using MQTTnet. - * Initial version. + * Updated to MQTTnet 2.7.5. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + - + - + - + - + - + diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 30bf138d..02f199f7 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -2,7 +2,7 @@ MQTTnet - 2.7.5 + 2.8.0 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,12 +10,9 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). - * [Client] Added new overloads for the message builders. -* [Core] Performance optimizations (thanks to @ israellot) -* [Core] Fixed a memory leak which was caused by not properly stopped async tasks. -* [Client] Fixed a deadlock when connecting to a not reachable server. -* [Server] Fixed a deadlock when reusing the same _ClientId_ while a _will message_ is used (thanks to @william-wps). -* [Server] Fixed wrong topic filter matching for filters like 'sport/+' which should _not_ match 'sport'. + * [Client] Received messages are now processed in the worker thread by default. Added a new setting for switching back to dedicated threads. +* [Server] Added support for other WebSocket sub protocol formats like mqttv-3.1.1 (thanks to @israellot). +* [Server] The takeover of an existing client sessions is now treated as a _clean_ disconnect of the previous client. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin diff --git a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs index 5a25712d..87b2038b 100644 --- a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs @@ -17,12 +17,12 @@ namespace MQTTnet.AspNetCore { string subprotocol = null; - if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues) - && requestedSubProtocolValues.Count > 0 - && requestedSubProtocolValues.Any(v => v.ToLower() == "mqtt") - ) + if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues)) { - subprotocol = "mqtt"; + // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc. + subprotocol = requestedSubProtocolValues + .OrderByDescending(p => p.Length) + .FirstOrDefault(p => p.ToLower().StartsWith("mqtt")); } var adapter = app.ApplicationServices.GetRequiredService(); diff --git a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs index 647f688b..aa873af0 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs @@ -6,17 +6,16 @@ namespace MQTTnet.Client public interface IMqttClientOptions { string ClientId { get; } - - IMqttClientCredentials Credentials { get; } bool CleanSession { get; } - MqttApplicationMessage WillMessage { get; } + IMqttClientCredentials Credentials { get; } + MqttProtocolVersion ProtocolVersion { get; } + IMqttClientChannelOptions ChannelOptions { get; } TimeSpan CommunicationTimeout { get; } TimeSpan KeepAlivePeriod { get; } TimeSpan? KeepAliveSendInterval { get; } + MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; } - MqttProtocolVersion ProtocolVersion { get; } - - IMqttClientChannelOptions ChannelOptions { get; } + MqttApplicationMessage WillMessage { get; } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs index dcbe8ae4..4c264f96 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs @@ -63,7 +63,7 @@ namespace MQTTnet.Client StartReceivingPackets(_cancellationTokenSource.Token); - var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false); + var connectResponse = await AuthenticateAsync(options.WillMessage, _cancellationTokenSource.Token).ConfigureAwait(false); _logger.Verbose("MQTT connection with server established."); _sendTracker.Restart(); @@ -94,7 +94,7 @@ namespace MQTTnet.Client { if (IsConnected && !_cancellationTokenSource.IsCancellationRequested) { - await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false); + await SendAsync(new MqttDisconnectPacket(), _cancellationTokenSource.Token).ConfigureAwait(false); } } finally @@ -115,7 +115,7 @@ namespace MQTTnet.Client TopicFilters = topicFilters.ToList() }; - var response = await SendAndReceiveAsync(subscribePacket).ConfigureAwait(false); + var response = await SendAndReceiveAsync(subscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false); if (response.SubscribeReturnCodes.Count != subscribePacket.TopicFilters.Count) { @@ -137,7 +137,7 @@ namespace MQTTnet.Client TopicFilters = topicFilters.ToList() }; - await SendAndReceiveAsync(unsubscribePacket).ConfigureAwait(false); + await SendAndReceiveAsync(unsubscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false); } public async Task PublishAsync(IEnumerable applicationMessages) @@ -154,7 +154,7 @@ namespace MQTTnet.Client case MqttQualityOfServiceLevel.AtMostOnce: { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await SendAsync(qosGroup.Cast().ToArray()).ConfigureAwait(false); + await SendAsync(qosGroup, _cancellationTokenSource.Token).ConfigureAwait(false); break; } case MqttQualityOfServiceLevel.AtLeastOnce: @@ -162,7 +162,7 @@ namespace MQTTnet.Client foreach (var publishPacket in qosGroup) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); - await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); + await SendAndReceiveAsync(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false); } break; @@ -173,13 +173,13 @@ namespace MQTTnet.Client { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); - var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); + var pubRecPacket = await SendAndReceiveAsync(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false); var pubRelPacket = new MqttPubRelPacket { PacketIdentifier = pubRecPacket.PacketIdentifier }; - await SendAndReceiveAsync(pubRelPacket).ConfigureAwait(false); + await SendAndReceiveAsync(pubRelPacket, _cancellationTokenSource.Token).ConfigureAwait(false); } break; @@ -200,7 +200,7 @@ namespace MQTTnet.Client _adapter?.Dispose(); } - private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage) + private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) { var connectPacket = new MqttConnectPacket { @@ -212,7 +212,7 @@ namespace MQTTnet.Client WillMessage = willApplicationMessage }; - var response = await SendAndReceiveAsync(connectPacket).ConfigureAwait(false); + var response = await SendAndReceiveAsync(connectPacket, _cancellationTokenSource.Token).ConfigureAwait(false); if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { throw new MqttConnectingFailedException(response.ConnectReturnCode); @@ -257,14 +257,12 @@ namespace MQTTnet.Client try { - if (_packetReceiverTask != null && _packetReceiverTask != sender) - { - _packetReceiverTask.Wait(); - } + await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); + await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false); if (_keepAliveMessageSenderTask != null && _keepAliveMessageSenderTask != sender) { - _keepAliveMessageSenderTask.Wait(); + await _keepAliveMessageSenderTask.ConfigureAwait(false); } if (_adapter != null) @@ -290,101 +288,19 @@ namespace MQTTnet.Client } } - private async Task ProcessReceivedPacketAsync(MqttBasePacket packet) - { - try - { - if (packet is MqttPublishPacket publishPacket) - { - await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false); - return; - } - - if (packet is MqttPingReqPacket) - { - await SendAsync(new MqttPingRespPacket()).ConfigureAwait(false); - return; - } - - if (packet is MqttDisconnectPacket) - { - await DisconnectAsync().ConfigureAwait(false); - return; - } - - if (packet is MqttPubRelPacket pubRelPacket) - { - await ProcessReceivedPubRelPacket(pubRelPacket).ConfigureAwait(false); - return; - } - - _packetDispatcher.Dispatch(packet); - } - catch (Exception exception) - { - _logger.Error(exception, "Unhandled exception while processing received packet."); - } - } - - private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) - { - try - { - var applicationMessage = publishPacket.ToApplicationMessage(); - ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); - } - catch (Exception exception) - { - _logger.Error(exception, "Unhandled exception while handling application message."); - } - } - - private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket) - { - if (_cancellationTokenSource.IsCancellationRequested) - { - return Task.FromResult(0); - } - - if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) - { - FireApplicationMessageReceivedEvent(publishPacket); - return Task.FromResult(0); - } - - if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) - { - FireApplicationMessageReceivedEvent(publishPacket); - return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); - } - - if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) - { - // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] - FireApplicationMessageReceivedEvent(publishPacket); - return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); - } - - throw new MqttCommunicationException("Received a not supported QoS level."); - } - - private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket) - { - var response = new MqttPubCompPacket - { - PacketIdentifier = pubRelPacket.PacketIdentifier - }; - - return SendAsync(response); - } - - private Task SendAsync(params MqttBasePacket[] packets) + private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken) { _sendTracker.Restart(); - return _adapter.SendPacketsAsync(_options.CommunicationTimeout, packets, _cancellationTokenSource.Token); + return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, new[] { packet }); } - private async Task SendAndReceiveAsync(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket + private Task SendAsync(IEnumerable packets, CancellationToken cancellationToken) + { + _sendTracker.Restart(); + return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, packets); + } + + private async Task SendAndReceiveAsync(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket { _sendTracker.Restart(); @@ -429,7 +345,7 @@ namespace MQTTnet.Client if (_sendTracker.Elapsed > keepAliveSendInterval) { - await SendAndReceiveAsync(new MqttPingReqPacket()).ConfigureAwait(false); + await SendAndReceiveAsync(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false); } await Task.Delay(keepAliveSendInterval, cancellationToken).ConfigureAwait(false); @@ -472,7 +388,19 @@ namespace MQTTnet.Client return; } - StartProcessReceivedPacket(packet, cancellationToken); + if (packet == null) + { + continue; + } + + if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.SingleThread) + { + await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); + } + else if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.DedicatedThread) + { + StartProcessReceivedPacketAsync(packet, cancellationToken); + } } } catch (Exception exception) @@ -498,9 +426,74 @@ namespace MQTTnet.Client } } - private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken) + private async Task ProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) { - Task.Run(() => ProcessReceivedPacketAsync(packet), cancellationToken); + try + { + if (packet is MqttPublishPacket publishPacket) + { + await ProcessReceivedPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); + return; + } + + if (packet is MqttPingReqPacket) + { + await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false); + return; + } + + if (packet is MqttDisconnectPacket) + { + await DisconnectAsync().ConfigureAwait(false); + return; + } + + if (packet is MqttPubRelPacket pubRelPacket) + { + await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false); + return; + } + + _packetDispatcher.Dispatch(packet); + } + catch (Exception exception) + { + _logger.Error(exception, "Unhandled exception while processing received packet."); + } + } + + private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) + { + FireApplicationMessageReceivedEvent(publishPacket); + return Task.FromResult(0); + } + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) + { + FireApplicationMessageReceivedEvent(publishPacket); + return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken); + } + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) + { + // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] + FireApplicationMessageReceivedEvent(publishPacket); + return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken); + } + + throw new MqttCommunicationException("Received a not supported QoS level."); + } + + private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) + { + var response = new MqttPubCompPacket + { + PacketIdentifier = pubRelPacket.PacketIdentifier + }; + + return SendAsync(response, cancellationToken); } private void StartReceivingPackets(CancellationToken cancellationToken) @@ -512,5 +505,46 @@ namespace MQTTnet.Client { _keepAliveMessageSenderTask = Task.Run(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken); } + + private void StartProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) + { +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Run(() => ProcessReceivedPacketAsync(packet, cancellationToken), cancellationToken); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + } + + private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) + { + try + { + var applicationMessage = publishPacket.ToApplicationMessage(); + ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); + } + catch (Exception exception) + { + _logger.Error(exception, "Unhandled exception while handling application message."); + } + } + + private static async Task WaitForTaskAsync(Task task, Task sender) + { + if (task == sender) + { + return; + } + + if (task.IsCanceled || task.IsCompleted || task.IsFaulted) + { + return; + } + + try + { + await task.ConfigureAwait(false); + } + catch (TaskCanceledException) + { + } + } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs index 9640a474..ae0517a6 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs @@ -5,22 +5,17 @@ namespace MQTTnet.Client { public class MqttClientOptions : IMqttClientOptions { - public MqttApplicationMessage WillMessage { get; set; } - public string ClientId { get; set; } = Guid.NewGuid().ToString("N"); - public bool CleanSession { get; set; } = true; - public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials(); - - public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15); - - public TimeSpan? KeepAliveSendInterval { get; set; } + public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; + public IMqttClientChannelOptions ChannelOptions { get; set; } public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10); + public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15); + public TimeSpan? KeepAliveSendInterval { get; set; } + public MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; set; } = MqttReceivedApplicationMessageProcessingMode.SingleThread; - public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; - - public IMqttClientChannelOptions ChannelOptions { get; set; } + public MqttApplicationMessage WillMessage { get; set; } } } diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs index 985cda7c..2eefb813 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs @@ -114,6 +114,13 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithReceivedApplicationMessageProcessingMode( + MqttReceivedApplicationMessageProcessingMode mode) + { + _options.ReceivedApplicationMessageProcessingMode = mode; + return this; + } + public IMqttClientOptions Build() { if (_tlsOptions != null) diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs new file mode 100644 index 00000000..651abe5b --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs @@ -0,0 +1,8 @@ +namespace MQTTnet.Client +{ + public enum MqttReceivedApplicationMessageProcessingMode + { + SingleThread, + DedicatedThread + } +} diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs new file mode 100644 index 00000000..19e9da8d --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs @@ -0,0 +1,8 @@ +namespace MQTTnet.Server +{ + public enum MqttClientDisconnectType + { + Clean, + NotClean + } +} diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs index d19ff463..8c98ac3f 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs @@ -127,7 +127,7 @@ namespace MQTTnet.Server if (!cancellationToken.IsCancellationRequested) { - await _clientSession.StopAsync().ConfigureAwait(false); + await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs index 83a684a2..3b51e9c9 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs @@ -101,7 +101,7 @@ namespace MQTTnet.Server return _wasCleanDisconnect; } - public Task StopAsync(bool wasCleanDisconnect = false) + public async Task StopAsync(MqttClientDisconnectType type) { try { @@ -110,7 +110,7 @@ namespace MQTTnet.Server return Task.FromResult(0); } - _wasCleanDisconnect = wasCleanDisconnect; + _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; _cancellationTokenSource?.Cancel(false); PendingMessagesQueue.WaitForCompletion(); @@ -119,7 +119,7 @@ namespace MQTTnet.Server var willMessage = _willMessage; _willMessage = null; // clear willmessage so it is send just once - if (_willMessage != null && !wasCleanDisconnect) + if (willMessage != null && !_wasCleanDisconnect) { _sessionsManager.StartDispatchApplicationMessage(this, willMessage); } @@ -186,7 +186,7 @@ namespace MQTTnet.Server private Task StopDueToKeepAliveTimeoutAsync() { _logger.Info("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId); - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) @@ -203,15 +203,18 @@ namespace MQTTnet.Server catch (OperationCanceledException) { } - catch (MqttCommunicationException exception) - { - _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); - await StopAsync().ConfigureAwait(false); - } catch (Exception exception) { - _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); - await StopAsync().ConfigureAwait(false); + if (exception is MqttCommunicationException) + { + _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); + } + else + { + _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); + } + + await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } } @@ -260,16 +263,16 @@ namespace MQTTnet.Server if (packet is MqttDisconnectPacket) { - return StopAsync(true); + return StopAsync(MqttClientDisconnectType.Clean); } if (packet is MqttConnectPacket) { - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } _logger.Warning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection topicFilters) @@ -288,7 +291,7 @@ namespace MQTTnet.Server if (subscribeResult.CloseConnection) { - await StopAsync().ConfigureAwait(false); + await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index 45854cf0..9302f0ae 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -120,7 +120,7 @@ namespace MQTTnet.Server { foreach (var session in _sessions) { - await session.Value.StopAsync().ConfigureAwait(false); + await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } _sessions.Clear(); @@ -232,7 +232,7 @@ namespace MQTTnet.Server { _sessions.Remove(connectPacket.ClientId); - await clientSession.StopAsync().ConfigureAwait(false); + await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false); clientSession.Dispose(); clientSession = null; diff --git a/MQTTnet.sln b/MQTTnet.sln index fab76088..a9bd6458 100644 --- a/MQTTnet.sln +++ b/MQTTnet.sln @@ -38,6 +38,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Extensions", "Extensions", EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.Rpc", "Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj", "{C444E9C8-95FA-430E-9126-274129DE16CD}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Benchmarks", "Tests\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{998D04DD-7CB0-45F5-A393-E2495C16399E}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -168,6 +170,22 @@ Global {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x64.Build.0 = Release|Any CPU {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x86.ActiveCfg = Release|Any CPU {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x86.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|Any CPU.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|ARM.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|ARM.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x64.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x64.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x86.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x86.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|Any CPU.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|Any CPU.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|ARM.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|ARM.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x64.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x64.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x86.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -180,6 +198,7 @@ Global {C6FF8AEA-0855-41EC-A1F3-AC262225BAB9} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4} {F10C4060-F7EE-4A83-919F-FF723E72F94A} = {32A630A7-2598-41D7-B625-204CD906F5FB} {C444E9C8-95FA-430E-9126-274129DE16CD} = {12816BCC-AF9E-44A9-9AE5-C246AF2A0587} + {998D04DD-7CB0-45F5-A393-E2495C16399E} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894} diff --git a/Tests/MQTTnet.Benchmarks/App.config b/Tests/MQTTnet.Benchmarks/App.config new file mode 100644 index 00000000..5a41bc2c --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/App.config @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj new file mode 100644 index 00000000..49773119 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -0,0 +1,169 @@ + + + + + Debug + AnyCPU + {998D04DD-7CB0-45F5-A393-E2495C16399E} + Exe + MQTTnet.Benchmarks + MQTTnet.Benchmarks + v4.6.2 + 512 + true + + + AnyCPU + true + full + false + bin\Debug\ + DEBUG;TRACE + prompt + 4 + + + AnyCPU + pdbonly + true + bin\Release\ + TRACE + prompt + 4 + + + + ..\..\packages\BenchmarkDotNet.0.10.14\lib\net46\BenchmarkDotNet.dll + + + ..\..\packages\BenchmarkDotNet.Core.0.10.14\lib\net46\BenchmarkDotNet.Core.dll + + + ..\..\packages\BenchmarkDotNet.Toolchains.Roslyn.0.10.14\lib\net46\BenchmarkDotNet.Toolchains.Roslyn.dll + + + ..\..\packages\Microsoft.CodeAnalysis.Common.2.6.1\lib\netstandard1.3\Microsoft.CodeAnalysis.dll + + + ..\..\packages\Microsoft.CodeAnalysis.CSharp.2.6.1\lib\netstandard1.3\Microsoft.CodeAnalysis.CSharp.dll + + + ..\..\packages\Microsoft.DotNet.InternalAbstractions.1.0.0\lib\net451\Microsoft.DotNet.InternalAbstractions.dll + + + ..\..\packages\Microsoft.DotNet.PlatformAbstractions.1.1.1\lib\net451\Microsoft.DotNet.PlatformAbstractions.dll + + + ..\..\packages\Microsoft.Win32.Registry.4.3.0\lib\net46\Microsoft.Win32.Registry.dll + + + + ..\..\packages\System.AppContext.4.3.0\lib\net46\System.AppContext.dll + True + + + ..\..\packages\System.Collections.Immutable.1.3.1\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll + True + + + + ..\..\packages\System.Console.4.3.0\lib\net46\System.Console.dll + + + + ..\..\packages\System.Diagnostics.FileVersionInfo.4.3.0\lib\net46\System.Diagnostics.FileVersionInfo.dll + + + ..\..\packages\System.Diagnostics.StackTrace.4.3.0\lib\net46\System.Diagnostics.StackTrace.dll + + + ..\..\packages\System.IO.Compression.4.3.0\lib\net46\System.IO.Compression.dll + True + + + ..\..\packages\System.IO.FileSystem.4.3.0\lib\net46\System.IO.FileSystem.dll + + + ..\..\packages\System.IO.FileSystem.Primitives.4.3.0\lib\net46\System.IO.FileSystem.Primitives.dll + + + + + ..\..\packages\System.Reflection.4.3.0\lib\net462\System.Reflection.dll + + + ..\..\packages\System.Reflection.Metadata.1.4.2\lib\portable-net45+win8\System.Reflection.Metadata.dll + + + ..\..\packages\System.Runtime.4.3.0\lib\net462\System.Runtime.dll + + + ..\..\packages\System.Runtime.Extensions.4.3.0\lib\net462\System.Runtime.Extensions.dll + + + ..\..\packages\System.Runtime.InteropServices.4.3.0\lib\net462\System.Runtime.InteropServices.dll + + + ..\..\packages\System.Security.Cryptography.Algorithms.4.3.0\lib\net461\System.Security.Cryptography.Algorithms.dll + + + ..\..\packages\System.Security.Cryptography.Encoding.4.3.0\lib\net46\System.Security.Cryptography.Encoding.dll + + + ..\..\packages\System.Security.Cryptography.Primitives.4.3.0\lib\net46\System.Security.Cryptography.Primitives.dll + + + ..\..\packages\System.Security.Cryptography.X509Certificates.4.3.0\lib\net461\System.Security.Cryptography.X509Certificates.dll + + + ..\..\packages\System.Text.Encoding.CodePages.4.3.0\lib\net46\System.Text.Encoding.CodePages.dll + + + ..\..\packages\System.Threading.Tasks.Extensions.4.3.0\lib\portable-net45+win8+wp8+wpa81\System.Threading.Tasks.Extensions.dll + + + ..\..\packages\System.Threading.Thread.4.3.0\lib\net46\System.Threading.Thread.dll + + + ..\..\packages\System.ValueTuple.4.3.0\lib\netstandard1.0\System.ValueTuple.dll + + + + + + + + + ..\..\packages\System.Xml.ReaderWriter.4.3.0\lib\net46\System.Xml.ReaderWriter.dll + + + ..\..\packages\System.Xml.XmlDocument.4.3.0\lib\net46\System.Xml.XmlDocument.dll + + + ..\..\packages\System.Xml.XPath.4.3.0\lib\net46\System.Xml.XPath.dll + + + ..\..\packages\System.Xml.XPath.XDocument.4.3.0\lib\net46\System.Xml.XPath.XDocument.dll + + + + + + + + + + + + + + + + + + {3587E506-55A2-4EB3-99C7-DC01E42D25D2} + MQTTnet.NetStandard + + + + \ No newline at end of file diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs new file mode 100644 index 00000000..53b71fd6 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs @@ -0,0 +1,47 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Attributes.Columns; +using BenchmarkDotNet.Attributes.Exporters; +using BenchmarkDotNet.Attributes.Jobs; +using MQTTnet.Client; +using MQTTnet.Server; + +namespace MQTTnet.Benchmarks +{ + [ClrJob] + [RPlotExporter, RankColumn] + public class MessageProcessingBenchmark + { + private IMqttServer _mqttServer; + private IMqttClient _mqttClient; + private MqttApplicationMessage _message; + + [GlobalSetup] + public void Setup() + { + var factory = new MqttFactory(); + _mqttServer = factory.CreateMqttServer(); + _mqttClient = factory.CreateMqttClient(); + + var serverOptions = new MqttServerOptionsBuilder().Build(); + _mqttServer.StartAsync(serverOptions).GetAwaiter().GetResult(); + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost").Build(); + + _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult(); + + _message = new MqttApplicationMessageBuilder() + .WithTopic("A") + .Build(); + } + + [Benchmark] + public void Send_10000_Messages() + { + for (var i = 0; i < 10000; i++) + { + _mqttClient.PublishAsync(_message).GetAwaiter().GetResult(); + } + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs new file mode 100644 index 00000000..965ddb55 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/Program.cs @@ -0,0 +1,14 @@ +using System; +using BenchmarkDotNet.Running; + +namespace MQTTnet.Benchmarks +{ + public static class Program + { + public static void Main(string[] args) + { + var summary = BenchmarkRunner.Run(); + Console.ReadLine(); + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Properties/AssemblyInfo.cs b/Tests/MQTTnet.Benchmarks/Properties/AssemblyInfo.cs new file mode 100644 index 00000000..4df83f6e --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/Properties/AssemblyInfo.cs @@ -0,0 +1,36 @@ +using System.Reflection; +using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; + +// General Information about an assembly is controlled through the following +// set of attributes. Change these attribute values to modify the information +// associated with an assembly. +[assembly: AssemblyTitle("MQTTnet.Benchmarks")] +[assembly: AssemblyDescription("")] +[assembly: AssemblyConfiguration("")] +[assembly: AssemblyCompany("")] +[assembly: AssemblyProduct("MQTTnet.Benchmarks")] +[assembly: AssemblyCopyright("Copyright © 2018")] +[assembly: AssemblyTrademark("")] +[assembly: AssemblyCulture("")] + +// Setting ComVisible to false makes the types in this assembly not visible +// to COM components. If you need to access a type in this assembly from +// COM, set the ComVisible attribute to true on that type. +[assembly: ComVisible(false)] + +// The following GUID is for the ID of the typelib if this project is exposed to COM +[assembly: Guid("998d04dd-7cb0-45f5-a393-e2495c16399e")] + +// Version information for an assembly consists of the following four values: +// +// Major Version +// Minor Version +// Build Number +// Revision +// +// You can specify all the values or you can default the Build and Revision Numbers +// by using the '*' as shown below: +// [assembly: AssemblyVersion("1.0.*")] +[assembly: AssemblyVersion("1.0.0.0")] +[assembly: AssemblyFileVersion("1.0.0.0")] diff --git a/Tests/MQTTnet.Benchmarks/packages.config b/Tests/MQTTnet.Benchmarks/packages.config new file mode 100644 index 00000000..bb930145 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/packages.config @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index 9e5f90ae..591cc87d 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -5,6 +5,7 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.FileProviders; +using MQTTnet; using MQTTnet.AspNetCore; using MQTTnet.Server;