From 0b5a751c5658c17a70f8983adb1912d0e3a84b61 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 15 Aug 2018 17:48:28 +0200 Subject: [PATCH 01/39] Add all methods of the factory to the factory interfaces. --- Source/MQTTnet/Client/IMqttClientFactory.cs | 4 ++++ Source/MQTTnet/Server/IMqttServerFactory.cs | 2 ++ 2 files changed, 6 insertions(+) diff --git a/Source/MQTTnet/Client/IMqttClientFactory.cs b/Source/MQTTnet/Client/IMqttClientFactory.cs index a15b499b..cd687afc 100644 --- a/Source/MQTTnet/Client/IMqttClientFactory.cs +++ b/Source/MQTTnet/Client/IMqttClientFactory.cs @@ -7,5 +7,9 @@ namespace MQTTnet.Client IMqttClient CreateMqttClient(); IMqttClient CreateMqttClient(IMqttNetLogger logger); + + IMqttClient CreateMqttClient(IMqttClientAdapterFactory adapterFactory); + + IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory adapterFactory); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/IMqttServerFactory.cs b/Source/MQTTnet/Server/IMqttServerFactory.cs index 6e399cae..7d23ed36 100644 --- a/Source/MQTTnet/Server/IMqttServerFactory.cs +++ b/Source/MQTTnet/Server/IMqttServerFactory.cs @@ -8,6 +8,8 @@ namespace MQTTnet.Server { IMqttServer CreateMqttServer(); + IMqttServer CreateMqttServer(IMqttNetLogger logger); + IMqttServer CreateMqttServer(IEnumerable adapters, IMqttNetLogger logger); } } \ No newline at end of file From 4cc9f12a03bf9359f0c2c094ec6807ab84367db2 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Tue, 21 Aug 2018 21:32:03 +0200 Subject: [PATCH 02/39] added AspNetMqttServerOptionsBuilder --- .../AspNetMqttServerOptionsBuilder.cs | 15 +++++++++++++++ .../ServiceCollectionExtensions.cs | 14 +++++++------- 2 files changed, 22 insertions(+), 7 deletions(-) create mode 100644 Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs diff --git a/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs b/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs new file mode 100644 index 00000000..e0ae3330 --- /dev/null +++ b/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs @@ -0,0 +1,15 @@ +using MQTTnet.Server; +using System; + +namespace MQTTnet.AspNetCore +{ + public class AspNetMqttServerOptionsBuilder : MqttServerOptionsBuilder + { + public AspNetMqttServerOptionsBuilder(IServiceProvider serviceProvider) + { + ServiceProvider = serviceProvider; + } + + public IServiceProvider ServiceProvider { get; } + } +} diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index b5d0440a..45a0df5c 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -21,13 +21,13 @@ namespace MQTTnet.AspNetCore return services; } - public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) - { - var builder = new MqttServerOptionsBuilder(); - - configure(builder); - - services.AddSingleton(builder.Build()); + public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) + { + services.AddSingleton(s => { + var builder = new AspNetMqttServerOptionsBuilder(s); + configure(builder); + return builder.Build(); + }); services.AddHostedMqttServer(); From 976c62d9c2def88e8220fa7b6af8ca93201c6fb0 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 25 Aug 2018 12:08:42 +0200 Subject: [PATCH 03/39] added back the old function and added new overload --- .../ServiceCollectionExtensions.cs | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index 45a0df5c..d9d9fdb8 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -20,9 +20,22 @@ namespace MQTTnet.AspNetCore return services; } - - public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) - { + + public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) + { + services.AddSingleton(s => { + var builder = new MqttServerOptionsBuilder(); + configure(builder); + return builder.Build(); + }); + + services.AddHostedMqttServer(); + + return services; + } + + public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action configure) + { services.AddSingleton(s => { var builder = new AspNetMqttServerOptionsBuilder(s); configure(builder); From 028a39ec25853fca020c0320f971fefa8b67e8d5 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 13 Sep 2018 09:19:15 +0200 Subject: [PATCH 04/39] Update docs. --- Build/MQTTnet.nuspec | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index b92aece6..c59abcac 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,15 +10,8 @@ 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). - * [Core] Performance optimizations. -* [Client] Added support for proxies when using web socket connections (thanks to PitySOFT). -* [Client] Refactored TLS parameter usage and added more parameters (thanks to PitySOFT). -* [Client] Fixed a deadlock when clients are disconnecting (thanks to @JanEggers). -* [ManagedClient] Unsubscriptions are now pushed before subscriptions to support re-subscribing (thanks to @cstichlberger). -* [Server] Fixed an issue in client keep alive checks (thanks to @jenscski). -* [Server] Changed the order of _ClientConnected_ and _ClientDisconnected_ events so that _ClientConnected_ is fired at first (thanks to @jenscski). -* [Server] Fixed an iussue which lets the server stop processing messages when using the application message interceptor (thanks to @alamsor). -* [Server] Fixed a deadlock when clients are disconnecting (thanks to @JanEggers). + * [Core] Added all factory methods to the factory interface. +* [Server] Added a new overload for configuring the ASP.net integration (thanks to @JanEggers). 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 From 682298b69fa2057086bf7fb26de6d456c49856bf Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 13 Sep 2018 09:29:26 +0200 Subject: [PATCH 05/39] Fix issue in pending messages handling for managed client. --- Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index a38bcc27..2b28deb9 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -226,7 +226,7 @@ namespace MQTTnet.Extensions.ManagedClient { try { - while (!cancellationToken.IsCancellationRequested) + while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected) { var message = _messageQueue.Take(cancellationToken); if (message == null) @@ -367,7 +367,6 @@ namespace MQTTnet.Extensions.ManagedClient } var cts = new CancellationTokenSource(); - _publishingCancellationToken = cts; Task.Factory.StartNew(() => PublishQueuedMessages(cts.Token), cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); From 39bc70cbf08c6253145efd97a877b745db7f1da2 Mon Sep 17 00:00:00 2001 From: Andre Crabb Date: Thu, 20 Sep 2018 07:17:35 -0400 Subject: [PATCH 06/39] * Cancel the TokenSource when disposing or nulling it out for the MqttClient --- Source/MQTTnet/Client/MqttClient.cs | 8 ++++---- Source/MQTTnet/Server/MqttClientSession.cs | 4 +++- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index ccedb5ae..bbe5ebd8 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -176,10 +176,12 @@ namespace MQTTnet.Client public void Dispose() { + _cancellationTokenSource?.Cancel (); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; _adapter?.Dispose(); + _adapter = null; } private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) @@ -238,10 +240,7 @@ namespace MQTTnet.Client } finally { - _adapter?.Dispose(); - _adapter = null; - _cancellationTokenSource?.Dispose(); - _cancellationTokenSource = null; + Dispose (); _cleanDisconnectInitiated = false; _logger.Info("Disconnected."); @@ -384,6 +383,7 @@ namespace MQTTnet.Client if (exception is OperationCanceledException) { + _logger.Info ("MQTT OperationCanceled exception while receiving packets."); } else if (exception is MqttCommunicationException) { diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 429fdb45..b2bb96b3 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -125,7 +125,7 @@ namespace MQTTnet.Server } _adapter = null; - + _cancellationTokenSource?.Cancel (); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; } @@ -240,7 +240,9 @@ namespace MQTTnet.Server { _pendingPacketsQueue?.Dispose(); + _cancellationTokenSource?.Cancel (); _cancellationTokenSource?.Dispose(); + _cancellationTokenSource = null; } private void ProcessReceivedPacket(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken) From b2e7a2e370ba28818b3bbdf2c36e8b3bbaf78c24 Mon Sep 17 00:00:00 2001 From: Andre Crabb Date: Fri, 21 Sep 2018 15:44:31 -0400 Subject: [PATCH 07/39] Change from logging `Info` to `Verbose` --- Source/MQTTnet/Client/MqttClient.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index bbe5ebd8..4d36ff6c 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -383,7 +383,7 @@ namespace MQTTnet.Client if (exception is OperationCanceledException) { - _logger.Info ("MQTT OperationCanceled exception while receiving packets."); + _logger.Verbose ("MQTT OperationCanceled exception while receiving packets."); } else if (exception is MqttCommunicationException) { From a272ed7daa2a0573195d36e4e20c7cceab9f899e Mon Sep 17 00:00:00 2001 From: Andre Crabb Date: Fri, 28 Sep 2018 11:00:18 -0400 Subject: [PATCH 08/39] * Change tabs to spaces. --- Source/MQTTnet/Client/MqttClient.cs | 4 ++-- Source/MQTTnet/Server/MqttClientSession.cs | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 4d36ff6c..63683193 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -176,12 +176,12 @@ namespace MQTTnet.Client public void Dispose() { - _cancellationTokenSource?.Cancel (); + _cancellationTokenSource?.Cancel (); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; _adapter?.Dispose(); - _adapter = null; + _adapter = null; } private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index b2bb96b3..95b15266 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -125,7 +125,7 @@ namespace MQTTnet.Server } _adapter = null; - _cancellationTokenSource?.Cancel (); + _cancellationTokenSource?.Cancel (); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; } @@ -240,9 +240,9 @@ namespace MQTTnet.Server { _pendingPacketsQueue?.Dispose(); - _cancellationTokenSource?.Cancel (); + _cancellationTokenSource?.Cancel (); _cancellationTokenSource?.Dispose(); - _cancellationTokenSource = null; + _cancellationTokenSource = null; } private void ProcessReceivedPacket(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken) From 00df7c9b7be801f6a5ad15d2f17121504244de13 Mon Sep 17 00:00:00 2001 From: Andre Crabb Date: Fri, 28 Sep 2018 11:01:07 -0400 Subject: [PATCH 09/39] * More change tabs to spaces. --- Source/MQTTnet/Client/MqttClient.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 63683193..4eb76d66 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -240,7 +240,7 @@ namespace MQTTnet.Client } finally { - Dispose (); + Dispose (); _cleanDisconnectInitiated = false; _logger.Info("Disconnected."); From 400c0f05adcb07b66c9f97b7a203a8b96ba6ce1b Mon Sep 17 00:00:00 2001 From: Andre Crabb Date: Fri, 28 Sep 2018 14:07:04 -0400 Subject: [PATCH 10/39] * Adds `false` param to Cancel call. --- Source/MQTTnet/Client/MqttClient.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 4eb76d66..f7093eb1 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -176,7 +176,7 @@ namespace MQTTnet.Client public void Dispose() { - _cancellationTokenSource?.Cancel (); + _cancellationTokenSource?.Cancel (false); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; From 70f4dc9e1066a4bdb984df2b25e55f0b18aee4aa Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 28 Sep 2018 22:51:05 +0200 Subject: [PATCH 11/39] Change log level of retained message changes. --- Source/MQTTnet/Server/MqttRetainedMessagesManager.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs index 86c321fc..6a515a11 100644 --- a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs @@ -93,7 +93,7 @@ namespace MQTTnet.Server if (applicationMessage.Payload?.Length == 0) { saveIsRequired = _messages.Remove(applicationMessage.Topic); - _logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); + _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); } else { @@ -111,7 +111,7 @@ namespace MQTTnet.Server } } - _logger.Info("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic); + _logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic); } } From 381c59d1e521c48bf4a449ebc505e2f5301a67d3 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 28 Sep 2018 22:51:16 +0200 Subject: [PATCH 12/39] Update readme. --- README.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 1a34eeed..fc2aceaf 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,11 @@

+
+

[![NuGet Badge](https://buildstats.info/nuget/MQTTnet)](https://www.nuget.org/packages/MQTTnet) -[![Build status](https://ci.appveyor.com/api/projects/status/ycit86voxfevm2aa/branch/master?svg=true)](https://ci.appveyor.com/project/chkr1011/mqttnet/branch/develop) +[![Build status](https://ci.appveyor.com/api/projects/status/ycit86voxfevm2aa/branch/master?svg=true)](https://ci.appveyor.com/project/chkr1011/mqttnet) [![BCH compliance](https://bettercodehub.com/edge/badge/chkr1011/MQTTnet?branch=master)](https://bettercodehub.com/) [![OpenCollective](https://opencollective.com/mqttnet/backers/badge.svg)](https://opencollective.com/mqttnet) [![OpenCollective](https://opencollective.com/mqttnet/sponsors/badge.svg)](https://opencollective.com/mqttnet) From 56a1530ceeeb2e481eceb069412602b2eeea82f0 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 29 Sep 2018 11:49:02 +0200 Subject: [PATCH 13/39] Add a method for clearing all retained messages at the server. --- Build/MQTTnet.nuspec | 2 ++ Source/MQTTnet/Server/IMqttServer.cs | 2 ++ Source/MQTTnet/Server/MqttRetainedMessagesManager.cs | 10 ++++++++++ Source/MQTTnet/Server/MqttServer.cs | 5 +++++ 4 files changed, 19 insertions(+) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index c59abcac..2dbe7310 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -11,7 +11,9 @@ false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). * [Core] Added all factory methods to the factory interface. +* [Core] Fixed an issue with cancellation token handling (thanks to @acrabb). * [Server] Added a new overload for configuring the ASP.net integration (thanks to @JanEggers). +* [Server] Added a method for clearing all retained messages. 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/Source/MQTTnet/Server/IMqttServer.cs b/Source/MQTTnet/Server/IMqttServer.cs index e724a65f..55ba710d 100644 --- a/Source/MQTTnet/Server/IMqttServer.cs +++ b/Source/MQTTnet/Server/IMqttServer.cs @@ -23,5 +23,7 @@ namespace MQTTnet.Server Task StartAsync(IMqttServerOptions options); Task StopAsync(); + + Task ClearRetainedMessagesAsync(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs index 6a515a11..c2c8eb0d 100644 --- a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs @@ -84,6 +84,16 @@ namespace MQTTnet.Server return retainedMessages; } + public Task ClearMessagesAsync() + { + lock (_messages) + { + _messages.Clear(); + } + + return _options.Storage.SaveRetainedMessagesAsync(new List()); + } + private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage) { var saveIsRequired = false; diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 6d532917..4086b18a 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -129,6 +129,11 @@ namespace MQTTnet.Server } } + public Task ClearRetainedMessagesAsync() + { + return _retainedMessagesManager?.ClearMessagesAsync(); + } + internal void OnClientConnected(string clientId) { _logger.Info("Client '{0}': Connected.", clientId); From ba62ca73064093fa4868a10c19c256bb344920c9 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 1 Oct 2018 21:36:15 +0200 Subject: [PATCH 14/39] Update docs. --- Build/MQTTnet.AspNetCore.nuspec | 2 +- Build/MQTTnet.Extensions.ManagedClient.nuspec | 2 +- Build/MQTTnet.Extensions.Rpc.nuspec | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 5be2f945..74de5d5c 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -15,7 +15,7 @@ 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.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index 99a58c4a..855947b5 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -15,7 +15,7 @@ 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 736f92da..45f1076d 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -15,7 +15,7 @@ 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 - + From 03bef4c15d5155b9f2334412a4b707f054a48d35 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 1 Oct 2018 21:50:54 +0200 Subject: [PATCH 15/39] Fix revocation checks in certificate validator. --- Source/MQTTnet/Implementations/MqttTcpChannel.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index bc10d65c..39283288 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -116,7 +116,7 @@ namespace MQTTnet.Implementations return true; } - if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.RevocationStatusUnknown)) + if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.OfflineRevocation)) { if (!_options.TlsOptions.IgnoreCertificateRevocationErrors) { From 0c2ab9b2319ce72aeb33dea8430708a6ab6d67c8 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 4 Oct 2018 20:53:02 +0200 Subject: [PATCH 16/39] Fix a deadlock when an exception is fired while connecting. --- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 6 ++- Source/MQTTnet/Client/MqttClient.cs | 42 ++++++++++-------- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 46 ++++++++++++++++++++ 3 files changed, 76 insertions(+), 18 deletions(-) diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 4e617330..13e66065 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -150,6 +150,9 @@ namespace MQTTnet.Adapter return packet; } + catch (OperationCanceledException) + { + } catch (Exception exception) { if (IsWrappedException(exception)) @@ -237,7 +240,8 @@ namespace MQTTnet.Adapter { if (exception is IOException && exception.InnerException is SocketException socketException) { - if (socketException.SocketErrorCode == SocketError.ConnectionAborted) + if (socketException.SocketErrorCode == SocketError.ConnectionAborted || + socketException.SocketErrorCode == SocketError.OperationAborted) { throw new OperationCanceledException(); } diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 575cc90f..0a408d08 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -54,12 +54,12 @@ namespace MQTTnet.Client try { - _cancellationTokenSource = new CancellationTokenSource(); - _disconnectReason = new TaskCompletionSource(); _options = options; _packetIdentifierProvider.Reset(); _packetDispatcher.Reset(); + _cancellationTokenSource = new CancellationTokenSource(); + _disconnectReason = new TaskCompletionSource(); _adapter = _adapterFactory.CreateClientAdapter(options, _logger); _logger.Verbose($"Trying to connect with server ({_options.ChannelOptions})."); @@ -87,10 +87,12 @@ namespace MQTTnet.Client catch (Exception exception) { _logger.Error(exception, "Error while connecting with server."); + if (_disconnectReason.TrySetException(exception)) { await DisconnectInternalAsync(null, exception).ConfigureAwait(false); } + throw; } } @@ -183,7 +185,7 @@ namespace MQTTnet.Client public void Dispose() { - _cancellationTokenSource?.Cancel (false); + _cancellationTokenSource?.Cancel(false); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; @@ -224,9 +226,10 @@ namespace MQTTnet.Client private async Task DisconnectInternalAsync(Task sender, Exception exception) { - InitiateDisconnect(); - var clientWasConnected = IsConnected; + + InitiateDisconnect(); + IsConnected = false; try @@ -247,7 +250,7 @@ namespace MQTTnet.Client } finally { - Dispose (); + Dispose(); _cleanDisconnectInitiated = false; _logger.Info("Disconnected."); @@ -261,16 +264,16 @@ namespace MQTTnet.Client { try { - if (_cancellationTokenSource == null || _cancellationTokenSource.IsCancellationRequested) + if (_cancellationTokenSource?.IsCancellationRequested == true) { return; } - _cancellationTokenSource.Cancel(false); + _cancellationTokenSource?.Cancel(false); } - catch (Exception adapterException) + catch (Exception exception) { - _logger.Warning(adapterException, "Error while initiating disconnect."); + _logger.Warning(exception, "Error while initiating disconnect."); } } } @@ -377,8 +380,10 @@ namespace MQTTnet.Client { while (!cancellationToken.IsCancellationRequested) { - var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); - if (packet != null) + var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken) + .ConfigureAwait(false); + + if (packet != null && !cancellationToken.IsCancellationRequested) { await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); } @@ -393,7 +398,6 @@ namespace MQTTnet.Client if (exception is OperationCanceledException) { - _logger.Verbose ("MQTT OperationCanceled exception while receiving packets."); } else if (exception is MqttCommunicationException) { @@ -492,16 +496,20 @@ namespace MQTTnet.Client private void StartReceivingPackets(CancellationToken cancellationToken) { - _packetReceiverTask = Task.Run( + _packetReceiverTask = Task.Factory.StartNew( () => ReceivePacketsAsync(cancellationToken), - cancellationToken); + cancellationToken, + TaskCreationOptions.LongRunning, + TaskScheduler.Default); } private void StartSendingKeepAliveMessages(CancellationToken cancellationToken) { - _keepAliveMessageSenderTask = Task.Run( + _keepAliveMessageSenderTask = Task.Factory.StartNew( () => SendKeepAliveMessagesAsync(cancellationToken), - cancellationToken); + cancellationToken, + TaskCreationOptions.LongRunning, + TaskScheduler.Default); } private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 0c34bdeb..e4e5b3b9 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -8,6 +8,7 @@ using System.Collections.Generic; using System.Text; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Adapter; using MQTTnet.Implementations; namespace MQTTnet.Core.Tests @@ -552,6 +553,51 @@ namespace MQTTnet.Core.Tests Assert.IsTrue(bodyIsMatching); } + [TestMethod] + public async Task MqttServer_ConnectionDenied() + { + var server = new MqttFactory().CreateMqttServer(); + var client = new MqttFactory().CreateMqttClient(); + + try + { + var options = new MqttServerOptionsBuilder().WithConnectionValidator(context => + { + context.ReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized; + }).Build(); + + await server.StartAsync(options); + + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost").Build(); + + try + { + await client.ConnectAsync(clientOptions); + Assert.Fail("An exception should be raised."); + } + catch (Exception exception) + { + if (exception is MqttConnectingFailedException) + { + + } + else + { + Assert.Fail("Wrong exception."); + } + } + } + finally + { + await client.DisconnectAsync(); + await server.StopAsync(); + + client.Dispose(); + } + } + [TestMethod] public async Task MqttServer_SameClientIdConnectDisconnectEventOrder() { From 6bee6a046946993a0f458d6e8b9c8626f327deff Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 4 Oct 2018 20:58:15 +0200 Subject: [PATCH 17/39] Update docs. --- Build/MQTTnet.nuspec | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 2dbe7310..fbcd355a 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,10 +10,7 @@ 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). - * [Core] Added all factory methods to the factory interface. -* [Core] Fixed an issue with cancellation token handling (thanks to @acrabb). -* [Server] Added a new overload for configuring the ASP.net integration (thanks to @JanEggers). -* [Server] Added a method for clearing all retained messages. + * [Client] Fixed a deadlock when an exception is fired while connecting (thanks to @malibVB). 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 From fb4f89b4120e79216df3c281ee676ec083a3c53b Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 4 Oct 2018 22:30:03 +0200 Subject: [PATCH 18/39] Replace TaskCompletionSource in Client with Interlocked gate. --- Source/MQTTnet/Client/MqttClient.cs | 17 +++++++++++------ Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 13 ++++++------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 0a408d08..05330b5d 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Client internal Task _keepAliveMessageSenderTask; private IMqttChannelAdapter _adapter; private bool _cleanDisconnectInitiated; - private TaskCompletionSource _disconnectReason; + private int _disconnectGate; public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger) { @@ -59,7 +59,7 @@ namespace MQTTnet.Client _packetDispatcher.Reset(); _cancellationTokenSource = new CancellationTokenSource(); - _disconnectReason = new TaskCompletionSource(); + _disconnectGate = 0; _adapter = _adapterFactory.CreateClientAdapter(options, _logger); _logger.Verbose($"Trying to connect with server ({_options.ChannelOptions})."); @@ -88,7 +88,7 @@ namespace MQTTnet.Client { _logger.Error(exception, "Error while connecting with server."); - if (_disconnectReason.TrySetException(exception)) + if (!DisconnectIsPending()) { await DisconnectInternalAsync(null, exception).ConfigureAwait(false); } @@ -110,7 +110,7 @@ namespace MQTTnet.Client } finally { - if (_disconnectReason.TrySetCanceled()) + if (!DisconnectIsPending()) { await DisconnectInternalAsync(null, null).ConfigureAwait(false); } @@ -361,7 +361,7 @@ namespace MQTTnet.Client _logger.Error(exception, "Unhandled exception while sending/receiving keep alive packets."); } - if (_disconnectReason.TrySetException(exception)) + if (!DisconnectIsPending()) { await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false); } @@ -410,7 +410,7 @@ namespace MQTTnet.Client _packetDispatcher.Dispatch(exception); - if (_disconnectReason.TrySetException(exception)) + if (!DisconnectIsPending()) { await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false); } @@ -545,5 +545,10 @@ namespace MQTTnet.Client { } } + + private bool DisconnectIsPending() + { + return Interlocked.CompareExchange(ref _disconnectGate, 1, 0) != 0; + } } } \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index e4e5b3b9..0404ccc0 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -669,25 +669,24 @@ namespace MQTTnet.Core.Tests MqttQualityOfServiceLevel filterQualityOfServiceLevel, int expectedReceivedMessagesCount) { - var serverAdapter = new TestMqttServerAdapter(); - var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger()); + var s = new MqttFactory().CreateMqttServer(); var receivedMessagesCount = 0; try { await s.StartAsync(new MqttServerOptions()); - var c1 = await serverAdapter.ConnectTestClient("c1"); - var c2 = await serverAdapter.ConnectTestClient("c2"); - + var c1 = new MqttFactory().CreateMqttClient(); c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - + await c1.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()); await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build()); + + var c2 = new MqttFactory().CreateMqttClient(); + await c2.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()); await c2.PublishAsync(builder => builder.WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel)); await Task.Delay(500); await c1.UnsubscribeAsync(topicFilter); - await Task.Delay(500); } finally From c8b3066883aafd2e3189e0daf7754aec3951dec2 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 6 Oct 2018 13:12:15 +0200 Subject: [PATCH 19/39] Update docs. --- Build/MQTTnet.AspNetCore.nuspec | 2 +- Build/MQTTnet.Extensions.ManagedClient.nuspec | 2 +- Build/MQTTnet.Extensions.Rpc.nuspec | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 5be2f945..cb1718f3 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -15,7 +15,7 @@ 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.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index 99a58c4a..e5d1184a 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -15,7 +15,7 @@ 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 736f92da..4075d654 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -15,7 +15,7 @@ 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 - + From b7ab29473b5a6902b6aba6cc0acda5d7cfca1cd9 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 7 Oct 2018 12:10:26 +0200 Subject: [PATCH 20/39] added ability to specify tls protocol version --- Source/MQTTnet/Implementations/MqttTcpServerListener.cs | 9 +++++++-- Source/MQTTnet/Server/MqttServerOptionsBuilder.cs | 7 +++++++ Source/MQTTnet/Server/MqttServerTlsTcpEndpointOptions.cs | 7 ++++++- 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index e3522979..e72c831d 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -20,8 +20,8 @@ namespace MQTTnet.Implementations private readonly CancellationToken _cancellationToken; private readonly AddressFamily _addressFamily; private readonly MqttServerTcpEndpointBaseOptions _options; + private readonly MqttServerTlsTcpEndpointOptions _tlsOptions; private readonly X509Certificate2 _tlsCertificate; - private Socket _socket; public MqttTcpServerListener( @@ -36,6 +36,11 @@ namespace MQTTnet.Implementations _tlsCertificate = tlsCertificate; _cancellationToken = cancellationToken; _logger = logger.CreateChildLogger(nameof(MqttTcpServerListener)); + + if (_options is MqttServerTlsTcpEndpointOptions tlsOptions) + { + _tlsOptions = tlsOptions; + } } public event EventHandler ClientAccepted; @@ -75,7 +80,7 @@ namespace MQTTnet.Implementations if (_tlsCertificate != null) { sslStream = new SslStream(new NetworkStream(clientSocket), false); - await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false); + await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false); } _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.", diff --git a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs index f54f379f..45400620 100644 --- a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs +++ b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs @@ -1,5 +1,6 @@ using System; using System.Net; +using System.Security.Authentication; namespace MQTTnet.Server { @@ -80,6 +81,12 @@ namespace MQTTnet.Server return this; } + public MqttServerOptionsBuilder WithEncryptionSslProtocol(SslProtocols value) + { + _options.TlsEndpointOptions.SslProtocol = value; + return this; + } + public MqttServerOptionsBuilder WithoutEncryptedEndpoint() { _options.TlsEndpointOptions.IsEnabled = false; diff --git a/Source/MQTTnet/Server/MqttServerTlsTcpEndpointOptions.cs b/Source/MQTTnet/Server/MqttServerTlsTcpEndpointOptions.cs index 76d6c862..88d2f86d 100644 --- a/Source/MQTTnet/Server/MqttServerTlsTcpEndpointOptions.cs +++ b/Source/MQTTnet/Server/MqttServerTlsTcpEndpointOptions.cs @@ -1,4 +1,6 @@ -namespace MQTTnet.Server +using System.Security.Authentication; + +namespace MQTTnet.Server { public class MqttServerTlsTcpEndpointOptions : MqttServerTcpEndpointBaseOptions { @@ -8,5 +10,8 @@ } public byte[] Certificate { get; set; } + + + public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12; } } From 583b32f17ae70e009dd95d702c24ee53b7749987 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 7 Oct 2018 13:24:07 +0200 Subject: [PATCH 21/39] updated release notes --- Build/MQTTnet.nuspec | 1 + 1 file changed, 1 insertion(+) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index fbcd355a..82536191 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -11,6 +11,7 @@ false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). * [Client] Fixed a deadlock when an exception is fired while connecting (thanks to @malibVB). + * [Server] Added ability to specify tls protocol version 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 From 73d33e2200b69a6867d6a83db9aed69ca385ecc9 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 7 Oct 2018 14:40:57 +0200 Subject: [PATCH 22/39] fixed #421 --- Build/MQTTnet.AspNetCore.nuspec | 3 +- .../MqttConnectionContext.cs | 3 +- .../MQTTnet.AspNetCore.Tests.csproj | 8 ++--- .../Mockups/ConnectionContextMockup.cs | 22 ++++++++++++++ .../Mockups/DuplexPipeMockup.cs | 16 ++++++++++ .../MqttConnectionContextTest.cs | 29 +++++++++++++++++++ .../MQTTnet.Core.Tests.csproj | 2 +- 7 files changed, 76 insertions(+), 7 deletions(-) create mode 100644 Tests/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs create mode 100644 Tests/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs create mode 100644 Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index cb1718f3..8809fac6 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -10,7 +10,8 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - * For more release notes please check the MQTTnet release notes. + * For more release notes please check the MQTTnet release notes. + * [Server] fixed cpu spike in case a client disconnectes (issue 421) 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/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 8a8469de..361eb6a1 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -1,6 +1,7 @@ using Microsoft.AspNetCore.Connections; using MQTTnet.Adapter; using MQTTnet.AspNetCore.Client.Tcp; +using MQTTnet.Exceptions; using MQTTnet.Packets; using MQTTnet.Serializer; using System; @@ -83,7 +84,7 @@ namespace MQTTnet.AspNetCore } else if (readResult.IsCompleted) { - break; + throw new MqttCommunicationException("Connection Aborted"); } } finally diff --git a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj index 77dfff97..f45b4fb5 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj +++ b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj @@ -1,4 +1,4 @@ - + netcoreapp2.1 @@ -7,9 +7,9 @@ - - - + + + diff --git a/Tests/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs b/Tests/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs new file mode 100644 index 00000000..296959ef --- /dev/null +++ b/Tests/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs @@ -0,0 +1,22 @@ +using System.Collections.Generic; +using System.IO.Pipelines; +using Microsoft.AspNetCore.Connections; +using Microsoft.AspNetCore.Http.Features; + +namespace MQTTnet.AspNetCore.Tests.Mockups +{ + public class ConnectionContextMockup : ConnectionContext + { + public override string ConnectionId { get; set; } + + public override IFeatureCollection Features { get; } + + public override IDictionary Items { get; set; } + public override IDuplexPipe Transport { get; set; } + + public ConnectionContextMockup() + { + //Transport = new DefaultConnectionContext + } + } +} diff --git a/Tests/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs b/Tests/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs new file mode 100644 index 00000000..1774f180 --- /dev/null +++ b/Tests/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs @@ -0,0 +1,16 @@ +using System.IO.Pipelines; + +namespace MQTTnet.AspNetCore.Tests.Mockups +{ + public class DuplexPipeMockup : IDuplexPipe + { + PipeReader IDuplexPipe.Input => Receive.Reader; + + PipeWriter IDuplexPipe.Output => Send.Writer; + + + public Pipe Receive { get; set; } = new Pipe(); + + public Pipe Send { get; set; } = new Pipe(); + } +} diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs new file mode 100644 index 00000000..7498015e --- /dev/null +++ b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs @@ -0,0 +1,29 @@ +using Microsoft.AspNetCore.Connections; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.AspNetCore.Tests.Mockups; +using MQTTnet.Exceptions; +using MQTTnet.Serializer; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.AspNetCore.Tests +{ + [TestClass] + public class MqttConnectionContextTest + { + [TestMethod] + public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted() + { + var serializer = new MqttPacketSerializer {}; + var pipe = new DuplexPipeMockup(); + var connection = new DefaultConnectionContext(); + connection.Transport = pipe; + var ctx = new MqttConnectionContext(serializer, connection); + + pipe.Receive.Writer.Complete(); + + await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(TimeSpan.FromSeconds(1), CancellationToken.None)); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj index 40697e94..ff820b29 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj +++ b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj @@ -9,7 +9,7 @@ - + From 4c76a898b6fb1c55531be173c1d7e087e79fc505 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 19 Oct 2018 23:55:02 +0200 Subject: [PATCH 23/39] Add new UnitTests. --- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 91 +++++++++++++++++-- .../MQTTnet.Core.Tests/RoundtripTimeTests.cs | 51 +++++++++++ .../TestServerExtensions.cs | 13 +-- 3 files changed, 143 insertions(+), 12 deletions(-) create mode 100644 Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 0404ccc0..fc403771 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -5,6 +5,7 @@ using MQTTnet.Protocol; using MQTTnet.Server; using System; using System.Collections.Generic; +using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; @@ -98,11 +99,11 @@ namespace MQTTnet.Core.Tests c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); - + await c2.PublishAsync(message); await Task.Delay(1000); Assert.AreEqual(0, receivedMessagesCount); - + var subscribeEventCalled = false; s.ClientSubscribedTopic += (_, e) => { @@ -116,7 +117,7 @@ namespace MQTTnet.Core.Tests await c2.PublishAsync(message); await Task.Delay(500); Assert.AreEqual(1, receivedMessagesCount); - + var unsubscribeEventCalled = false; s.ClientUnsubscribedTopic += (_, e) => { @@ -234,7 +235,7 @@ namespace MQTTnet.Core.Tests await c1.PublishAsync(message); } } - + [TestMethod] public async Task MqttServer_ShutdownDisconnectsClientsGracefully() { @@ -303,6 +304,57 @@ namespace MQTTnet.Core.Tests Assert.AreEqual(clientConnectedCalled, clientDisconnectedCalled); } + [TestMethod] + public async Task MqttServer_LotsOfRetainedMessages() + { + const int ClientCount = 100; + + var server = new MqttFactory().CreateMqttServer(); + try + { + await server.StartAsync(new MqttServerOptionsBuilder().Build()); + + Parallel.For( + 0, + ClientCount, + new ParallelOptions { MaxDegreeOfParallelism = 10 }, + i => + { + using (var client = new MqttFactory().CreateMqttClient()) + { + client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()) + .GetAwaiter().GetResult(); + + for (var j = 0; j < 10; j++) + { + // Clear retained message. + client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("r" + i) + .WithPayload(new byte[0]).WithRetainFlag().Build()).GetAwaiter().GetResult(); + + // Set retained message. + client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("r" + i) + .WithPayload("value" + j).WithRetainFlag().Build()).GetAwaiter().GetResult(); + } + + client.DisconnectAsync().GetAwaiter().GetResult(); + } + }); + + var retainedMessages = server.GetRetainedMessages(); + + Assert.AreEqual(ClientCount, retainedMessages.Count); + + for (var i = 0; i < ClientCount; i++) + { + Assert.IsTrue(retainedMessages.Any(m => m.Topic == "r" + i)); + } + } + finally + { + await server.StopAsync(); + } + } + [TestMethod] public async Task MqttServer_RetainedMessagesFlow() { @@ -568,7 +620,7 @@ namespace MQTTnet.Core.Tests await server.StartAsync(options); - + var clientOptions = new MqttClientOptionsBuilder() .WithTcpServer("localhost").Build(); @@ -593,7 +645,7 @@ namespace MQTTnet.Core.Tests { await client.DisconnectAsync(); await server.StopAsync(); - + client.Dispose(); } } @@ -646,6 +698,33 @@ namespace MQTTnet.Core.Tests Assert.IsFalse(connecteCalledBeforeConnectedClients, "ClientConnected was called before ClientDisconnect was called"); } + + [TestMethod] + public async Task MqttServer_StopAndRestart() + { + var server = new MqttFactory().CreateMqttServer(); + await server.StartAsync(new MqttServerOptions()); + + var client = new MqttFactory().CreateMqttClient(); + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()); + await server.StopAsync(); + + try + { + var client2 = new MqttFactory().CreateMqttClient(); + await client2.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()); + + Assert.Fail("Connecting should fail."); + } + catch (Exception) + { + } + + await server.StartAsync(new MqttServerOptions()); + var client3 = new MqttFactory().CreateMqttClient(); + await client3.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()); + } + private class TestStorage : IMqttServerStorage { public IList Messages = new List(); diff --git a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs new file mode 100644 index 00000000..7b5ad77a --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs @@ -0,0 +1,51 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Server; + +namespace MQTTnet.Core.Tests +{ + [TestClass] + public class RoundtripTimeTests + { + [TestMethod] + public async Task Round_Trip_Time() + { + var factory = new MqttFactory(); + var server = factory.CreateMqttServer(); + var receiverClient = factory.CreateMqttClient(); + var senderClient = factory.CreateMqttClient(); + + await server.StartAsync(new MqttServerOptionsBuilder().Build()); + + await receiverClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()); + await receiverClient.SubscribeAsync("#"); + + await senderClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build()); + + TaskCompletionSource response = null; + + receiverClient.ApplicationMessageReceived += (sender, args) => + { + response?.SetResult(args.ApplicationMessage.ConvertPayloadToString()); + }; + + var times = new List(); + var stopwatch = Stopwatch.StartNew(); + + for (var i = 0; i < 100; i++) + { + response = new TaskCompletionSource(); + await senderClient.PublishAsync("test", DateTime.UtcNow.Ticks.ToString()); + response.Task.GetAwaiter().GetResult(); + + stopwatch.Stop(); + times.Add(stopwatch.Elapsed); + stopwatch.Restart(); + } + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/TestServerExtensions.cs b/Tests/MQTTnet.Core.Tests/TestServerExtensions.cs index cbb278ca..7636b1a2 100644 --- a/Tests/MQTTnet.Core.Tests/TestServerExtensions.cs +++ b/Tests/MQTTnet.Core.Tests/TestServerExtensions.cs @@ -11,19 +11,20 @@ namespace MQTTnet.Core.Tests /// publishes a message with a client and waits in the server until a message with the same topic is received /// /// - public static async Task PublishAndWaitForAsync(this IMqttClient client, IMqttServer server, MqttApplicationMessage message) + public static async Task PublishAndWaitForAsync(this IMqttClient client, IMqttServer server, MqttApplicationMessage message) { var tcs = new TaskCompletionSource(); - EventHandler handler = (sender, args) => + void Handler(object sender, MqttApplicationMessageReceivedEventArgs args) { if (args.ApplicationMessage.Topic == message.Topic) { tcs.SetResult(true); } - }; - server.ApplicationMessageReceived += handler; - + } + + server.ApplicationMessageReceived += Handler; + try { await client.PublishAsync(message).ConfigureAwait(false); @@ -31,7 +32,7 @@ namespace MQTTnet.Core.Tests } finally { - server.ApplicationMessageReceived -= handler; + server.ApplicationMessageReceived -= Handler; } } } From aad3ab903f87b429d080bbab0a145b7fa6c5105b Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 19 Oct 2018 23:56:32 +0200 Subject: [PATCH 24/39] Expose retained messages to server interface. --- Source/MQTTnet/Server/IMqttServer.cs | 8 +- .../Server/MqttClientSessionsManager.cs | 4 +- .../Server/MqttRetainedMessagesManager.cs | 103 ++++++++++-------- Source/MQTTnet/Server/MqttServer.cs | 12 +- 4 files changed, 75 insertions(+), 52 deletions(-) diff --git a/Source/MQTTnet/Server/IMqttServer.cs b/Source/MQTTnet/Server/IMqttServer.cs index 55ba710d..3a45973b 100644 --- a/Source/MQTTnet/Server/IMqttServer.cs +++ b/Source/MQTTnet/Server/IMqttServer.cs @@ -16,14 +16,18 @@ namespace MQTTnet.Server IMqttServerOptions Options { get; } + [Obsolete("This method is no longer async. Use the not async method.")] Task> GetClientSessionsStatusAsync(); + IList GetClientSessionsStatus(); + + IList GetRetainedMessages(); + Task ClearRetainedMessagesAsync(); + Task SubscribeAsync(string clientId, IList topicFilters); Task UnsubscribeAsync(string clientId, IList topicFilters); Task StartAsync(IMqttServerOptions options); Task StopAsync(); - - Task ClearRetainedMessagesAsync(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index d8f4f141..400045b4 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -65,7 +65,7 @@ namespace MQTTnet.Server return Task.Run(() => RunSession(clientAdapter, _cancellationToken), _cancellationToken); } - public Task> GetClientStatusAsync() + public IList GetClientStatus() { var result = new List(); @@ -77,7 +77,7 @@ namespace MQTTnet.Server result.Add(status); } - return Task.FromResult((IList)result); + return result; } public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) diff --git a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs index c2c8eb0d..bf5de984 100644 --- a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs @@ -52,7 +52,51 @@ namespace MQTTnet.Server try { - await HandleMessageInternalAsync(clientId, applicationMessage).ConfigureAwait(false); + List messagesForSave = null; + lock (_messages) + { + var saveIsRequired = false; + + if (applicationMessage.Payload?.Length == 0) + { + saveIsRequired = _messages.Remove(applicationMessage.Topic); + _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); + } + else + { + if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage)) + { + _messages[applicationMessage.Topic] = applicationMessage; + saveIsRequired = true; + } + else + { + if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0])) + { + _messages[applicationMessage.Topic] = applicationMessage; + saveIsRequired = true; + } + } + + _logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic); + } + + if (saveIsRequired) + { + messagesForSave = new List(_messages.Values); + } + } + + if (messagesForSave == null) + { + _logger.Verbose("Skipped saving retained messages because no changes were detected."); + return; + } + + if (_options.Storage != null) + { + await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false); + } } catch (Exception exception) { @@ -84,6 +128,14 @@ namespace MQTTnet.Server return retainedMessages; } + public IList GetMessages() + { + lock (_messages) + { + return _messages.Values.ToList(); + } + } + public Task ClearMessagesAsync() { lock (_messages) @@ -91,55 +143,12 @@ namespace MQTTnet.Server _messages.Clear(); } - return _options.Storage.SaveRetainedMessagesAsync(new List()); - } - - private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage) - { - var saveIsRequired = false; - - lock (_messages) + if (_options.Storage != null) { - if (applicationMessage.Payload?.Length == 0) - { - saveIsRequired = _messages.Remove(applicationMessage.Topic); - _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); - } - else - { - if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage)) - { - _messages[applicationMessage.Topic] = applicationMessage; - saveIsRequired = true; - } - else - { - if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0])) - { - _messages[applicationMessage.Topic] = applicationMessage; - saveIsRequired = true; - } - } - - _logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic); - } + return _options.Storage.SaveRetainedMessagesAsync(new List()); } - if (!saveIsRequired) - { - _logger.Verbose("Skipped saving retained messages because no changes were detected."); - } - - if (saveIsRequired && _options.Storage != null) - { - List messages; - lock (_messages) - { - messages = _messages.Values.ToList(); - } - - await _options.Storage.SaveRetainedMessagesAsync(messages).ConfigureAwait(false); - } + return Task.FromResult((object)null); } } } diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 4086b18a..a960ab00 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -41,7 +41,17 @@ namespace MQTTnet.Server public Task> GetClientSessionsStatusAsync() { - return _clientSessionsManager.GetClientStatusAsync(); + return Task.FromResult(_clientSessionsManager.GetClientStatus()); + } + + public IList GetClientSessionsStatus() + { + return _clientSessionsManager.GetClientStatus(); + } + + public IList GetRetainedMessages() + { + return _retainedMessagesManager.GetMessages(); } public Task SubscribeAsync(string clientId, IList topicFilters) From c9a6155834044328893f90c6b872b5c5efc7ab72 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 19 Oct 2018 23:57:09 +0200 Subject: [PATCH 25/39] Update docs. --- Build/MQTTnet.nuspec | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index fbcd355a..6febc5ce 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,7 +10,7 @@ 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] Fixed a deadlock when an exception is fired while connecting (thanks to @malibVB). + * [Server] Added new method which exposes all retained messages. 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 From 5abee2512fb42ae642b80e3855042e5c8bcaed5c Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 20 Oct 2018 00:32:53 +0200 Subject: [PATCH 26/39] Update nuget packages. --- Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj | 6 +++--- Source/MQTTnet/MQTTnet.csproj | 2 +- .../MQTTnet.AspNetCore.Tests.csproj | 6 +++--- Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs | 2 -- Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj | 6 +++--- Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs | 3 --- Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs | 2 -- Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs | 2 -- Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj | 4 ++-- .../MQTTnet.TestApp.AspNetCore2.csproj | 6 ++++-- .../MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj | 6 +++--- .../MQTTnet.TestApp.UniversalWindows.csproj | 4 ++-- 12 files changed, 21 insertions(+), 28 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj index dd3a4d55..dc477f76 100644 --- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj +++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj @@ -14,9 +14,9 @@ - - - + + + diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 30eace10..3d3d80da 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -54,7 +54,7 @@ - + diff --git a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj index 77dfff97..55f60d91 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj +++ b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj @@ -7,9 +7,9 @@ - - - + + + diff --git a/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs b/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs index 2dce560a..0039434e 100644 --- a/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs @@ -1,6 +1,4 @@ using BenchmarkDotNet.Attributes; -using BenchmarkDotNet.Attributes.Exporters; -using BenchmarkDotNet.Attributes.Jobs; using MQTTnet.Diagnostics; namespace MQTTnet.Benchmarks diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj index 2a6b91a5..33f08667 100644 --- a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -8,9 +8,9 @@ - - - + + + diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs index f6ff61a3..5e99201c 100644 --- a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs @@ -1,7 +1,4 @@ using BenchmarkDotNet.Attributes; -using BenchmarkDotNet.Attributes.Columns; -using BenchmarkDotNet.Attributes.Exporters; -using BenchmarkDotNet.Attributes.Jobs; using MQTTnet.Client; using MQTTnet.Server; diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 9b855805..ed18df1d 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -2,8 +2,6 @@ using MQTTnet.Packets; using MQTTnet.Serializer; using MQTTnet.Internal; -using BenchmarkDotNet.Attributes.Jobs; -using BenchmarkDotNet.Attributes.Exporters; using System; using System.Threading; using System.Threading.Tasks; diff --git a/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs b/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs index 6d9a0321..bbce6305 100644 --- a/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs @@ -1,6 +1,4 @@ using BenchmarkDotNet.Attributes; -using BenchmarkDotNet.Attributes.Exporters; -using BenchmarkDotNet.Attributes.Jobs; using MQTTnet.Server; using System; diff --git a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj index 40697e94..20cd0452 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj +++ b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj @@ -3,13 +3,13 @@ Exe Full - netcoreapp2.0 + netcoreapp2.1 - + diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj index a216d728..a23f309b 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj +++ b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj @@ -10,8 +10,10 @@ - - + + + + diff --git a/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj b/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj index 2a738a08..af5f321c 100644 --- a/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj +++ b/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj @@ -3,11 +3,11 @@ Exe Full - netcoreapp2.0;net452;net461 + netcoreapp2.1;net452;net461 - - RELEASE;NETCOREAPP2_0 + + RELEASE;NETCOREAPP2_1 diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj index a6f4566b..f8583aa1 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj @@ -142,10 +142,10 @@ - 6.1.5 + 6.1.9 - 3.0.0 + 4.0.0 From ca7952ab33f2a3815b1fc5f36d05c75c40e0b20e Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 29 Oct 2018 20:35:22 +0100 Subject: [PATCH 27/39] Remove wrong setter. --- Build/MQTTnet.nuspec | 1 + Source/MQTTnet/Server/IMqttServerOptions.cs | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 6febc5ce..da658a71 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -11,6 +11,7 @@ false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). * [Server] Added new method which exposes all retained messages. +* [Server] Removed (wrong) setter from the server options interface. 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/Source/MQTTnet/Server/IMqttServerOptions.cs b/Source/MQTTnet/Server/IMqttServerOptions.cs index 2ae4c9fa..f9334fd7 100644 --- a/Source/MQTTnet/Server/IMqttServerOptions.cs +++ b/Source/MQTTnet/Server/IMqttServerOptions.cs @@ -14,7 +14,7 @@ namespace MQTTnet.Server Action ConnectionValidator { get; } Action SubscriptionInterceptor { get; } Action ApplicationMessageInterceptor { get; } - Action ClientMessageQueueInterceptor { get; set; } + Action ClientMessageQueueInterceptor { get; } MqttServerTcpEndpointOptions DefaultEndpointOptions { get; } MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; } From 28efbcdb3d09dd37cf40d659c11f991666cb2d3c Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 29 Oct 2018 21:45:01 +0100 Subject: [PATCH 28/39] Add max pending messages options for ManagedClient. --- Build/MQTTnet.nuspec | 2 + README.md | 2 +- .../ApplicationMessageSkippedEventArgs.cs | 14 +++ .../IManagedMqttClient.cs | 1 + .../IManagedMqttClientOptions.cs | 5 + .../ManagedMqttClient.cs | 52 ++++++--- .../ManagedMqttClientOptions.cs | 5 + .../ManagedMqttClientOptionsBuilder.cs | 13 +++ Source/MQTTnet/Internal/BlockingQueue.cs | 77 +++++++++++++ .../MQTTnet.Core.Tests/BlockingQueueTests.cs | 108 ++++++++++++++++++ .../MQTTnet.Core.Tests.csproj | 1 + .../ManagedMqttClientTests.cs | 38 ++++++ Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 2 + .../MQTTnet.Core.Tests/RoundtripTimeTests.cs | 4 +- 14 files changed, 309 insertions(+), 15 deletions(-) create mode 100644 Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs create mode 100644 Source/MQTTnet/Internal/BlockingQueue.cs create mode 100644 Tests/MQTTnet.Core.Tests/BlockingQueueTests.cs create mode 100644 Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index da658a71..71f17dbe 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -12,6 +12,8 @@ MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). * [Server] Added new method which exposes all retained messages. * [Server] Removed (wrong) setter from the server options interface. +* [ManagedClient] Added max pending messages count option. +* [ManagedClient] Add pending messages overflow strategy option. 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/README.md b/README.md index fc2aceaf..cb405dd4 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov * Performance optimized (processing ~70.000 messages / second)* * Interfaces included for mocking and testing * Access to internal trace messages -* Unit tested (~90 tests) +* Unit tested (~100 tests) \* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_. diff --git a/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs new file mode 100644 index 00000000..0f1613d4 --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs @@ -0,0 +1,14 @@ +using System; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class ApplicationMessageSkippedEventArgs : EventArgs + { + public ApplicationMessageSkippedEventArgs(ManagedMqttApplicationMessage applicationMessage) + { + ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); + } + + public ManagedMqttApplicationMessage ApplicationMessage { get; } + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index b5a6c50b..d0035849 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -15,6 +15,7 @@ namespace MQTTnet.Extensions.ManagedClient event EventHandler Disconnected; event EventHandler ApplicationMessageProcessed; + event EventHandler ApplicationMessageSkipped; event EventHandler ConnectingFailed; event EventHandler SynchronizingSubscriptionsFailed; diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs index 8d48babd..f59aca7c 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Client; +using MQTTnet.Server; namespace MQTTnet.Extensions.ManagedClient { @@ -12,5 +13,9 @@ namespace MQTTnet.Extensions.ManagedClient TimeSpan ConnectionCheckInterval { get; } IManagedMqttClientStorage Storage { get; } + + int MaxPendingMessages { get; } + + MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 2b28deb9..16d8e9b5 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -7,13 +6,17 @@ using System.Threading.Tasks; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; +using MQTTnet.Internal; using MQTTnet.Protocol; +using MQTTnet.Server; +using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs; +using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs; namespace MQTTnet.Extensions.ManagedClient { public class ManagedMqttClient : IManagedMqttClient { - private readonly BlockingCollection _messageQueue = new BlockingCollection(); + private readonly BlockingQueue _messageQueue = new BlockingQueue(); private readonly Dictionary _subscriptions = new Dictionary(); private readonly HashSet _unsubscriptions = new HashSet(); @@ -50,6 +53,7 @@ namespace MQTTnet.Extensions.ManagedClient public event EventHandler ApplicationMessageReceived; public event EventHandler ApplicationMessageProcessed; + public event EventHandler ApplicationMessageSkipped; public event EventHandler ConnectingFailed; public event EventHandler SynchronizingSubscriptionsFailed; @@ -75,7 +79,7 @@ namespace MQTTnet.Extensions.ManagedClient foreach (var message in messages) { - _messageQueue.Add(message); + _messageQueue.Enqueue(message); } } @@ -93,10 +97,7 @@ namespace MQTTnet.Extensions.ManagedClient StopPublishing(); StopMaintainingConnection(); - while (_messageQueue.Any()) - { - _messageQueue.Take(); - } + _messageQueue.Clear(); return Task.FromResult(0); } @@ -112,12 +113,38 @@ namespace MQTTnet.Extensions.ManagedClient { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - if (_storageManager != null) + ManagedMqttApplicationMessage skippedMessage = null; + lock (_messageQueue) { - await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false); + if (_messageQueue.Count > _options.MaxPendingMessages) + { + if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) + { + _logger.Verbose("Skipping publish of new application message because internal queue is full."); + ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(applicationMessage)); + return; + } + + if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage) + { + skippedMessage = _messageQueue.RemoveFirst(); + _logger.Verbose("Removed oldest application message from internal queue because it is full."); + ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(skippedMessage)); + } + } + + _messageQueue.Enqueue(applicationMessage); } - _messageQueue.Add(applicationMessage); + if (_storageManager != null) + { + if (skippedMessage != null) + { + await _storageManager.RemoveAsync(skippedMessage).ConfigureAwait(false); + } + + await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false); + } } public Task SubscribeAsync(IEnumerable topicFilters) @@ -157,7 +184,6 @@ namespace MQTTnet.Extensions.ManagedClient public void Dispose() { - _messageQueue?.Dispose(); _connectionCancellationToken?.Dispose(); _publishingCancellationToken?.Dispose(); } @@ -228,7 +254,7 @@ namespace MQTTnet.Extensions.ManagedClient { while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected) { - var message = _messageQueue.Take(cancellationToken); + var message = _messageQueue.Dequeue(); if (message == null) { continue; @@ -268,7 +294,7 @@ namespace MQTTnet.Extensions.ManagedClient if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - _messageQueue.Add(message); + _messageQueue.Enqueue(message); } } catch (Exception exception) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs index cecf2274..cba801bb 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Client; +using MQTTnet.Server; namespace MQTTnet.Extensions.ManagedClient { @@ -12,5 +13,9 @@ namespace MQTTnet.Extensions.ManagedClient public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1); public IManagedMqttClientStorage Storage { get; set; } + + public int MaxPendingMessages { get; set; } = int.MaxValue; + + public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropNewMessage; } } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs index ef774c59..3ea5b050 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Client; +using MQTTnet.Server; namespace MQTTnet.Extensions.ManagedClient { @@ -8,6 +9,18 @@ namespace MQTTnet.Extensions.ManagedClient private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions(); private MqttClientOptionsBuilder _clientOptionsBuilder; + public ManagedMqttClientOptionsBuilder WithMaxPendingMessages(int value) + { + _options.MaxPendingMessages = value; + return this; + } + + public ManagedMqttClientOptionsBuilder WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy value) + { + _options.PendingMessagesOverflowStrategy = value; + return this; + } + public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value) { _options.AutoReconnectDelay = value; diff --git a/Source/MQTTnet/Internal/BlockingQueue.cs b/Source/MQTTnet/Internal/BlockingQueue.cs new file mode 100644 index 00000000..ae6cc985 --- /dev/null +++ b/Source/MQTTnet/Internal/BlockingQueue.cs @@ -0,0 +1,77 @@ +using System; +using System.Collections.Generic; +using System.Threading; + +namespace MQTTnet.Internal +{ + public class BlockingQueue + { + private readonly object _syncRoot = new object(); + private readonly LinkedList _items = new LinkedList(); + private readonly ManualResetEvent _gate = new ManualResetEvent(false); + + public int Count + { + get + { + lock (_syncRoot) + { + return _items.Count; + } + } + } + + public void Enqueue(TItem item) + { + if (item == null) throw new ArgumentNullException(nameof(item)); + + lock (_syncRoot) + { + _items.AddLast(item); + _gate.Set(); + } + } + + public TItem Dequeue() + { + while (true) + { + lock (_syncRoot) + { + if (_items.Count > 0) + { + var item = _items.First.Value; + _items.RemoveFirst(); + + return item; + } + + if (_items.Count == 0) + { + _gate.Reset(); + } + } + + _gate.WaitOne(); + } + } + + public TItem RemoveFirst() + { + lock (_syncRoot) + { + var item = _items.First; + _items.RemoveFirst(); + return item.Value; + } + } + + public void Clear() + { + lock (_syncRoot) + { + _items.Clear(); + } + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/BlockingQueueTests.cs b/Tests/MQTTnet.Core.Tests/BlockingQueueTests.cs new file mode 100644 index 00000000..80e0d724 --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/BlockingQueueTests.cs @@ -0,0 +1,108 @@ +using System.Threading; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Internal; + +namespace MQTTnet.Core.Tests +{ + [TestClass] + public class BlockingQueueTests + { + [TestMethod] + public void Preserve_Order() + { + var queue = new BlockingQueue(); + queue.Enqueue("a"); + queue.Enqueue("b"); + queue.Enqueue("c"); + + Assert.AreEqual(3, queue.Count); + + Assert.AreEqual("a", queue.Dequeue()); + Assert.AreEqual("b", queue.Dequeue()); + Assert.AreEqual("c", queue.Dequeue()); + } + + [TestMethod] + public void Remove_First_Items() + { + var queue = new BlockingQueue(); + queue.Enqueue("a"); + queue.Enqueue("b"); + queue.Enqueue("c"); + + Assert.AreEqual("a", queue.RemoveFirst()); + Assert.AreEqual("b", queue.RemoveFirst()); + + Assert.AreEqual(1, queue.Count); + + Assert.AreEqual("c", queue.Dequeue()); + } + + [TestMethod] + public void Clear_Items() + { + var queue = new BlockingQueue(); + queue.Enqueue("a"); + queue.Enqueue("b"); + queue.Enqueue("c"); + + Assert.AreEqual(3, queue.Count); + + queue.Clear(); + + Assert.AreEqual(0, queue.Count); + } + + [TestMethod] + public async Task Wait_For_Item() + { + var queue = new BlockingQueue(); + + string item = null; +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Run(() => + { + item = queue.Dequeue(); + }); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + + await Task.Delay(100); + + Assert.AreEqual(0, queue.Count); + Assert.AreEqual(null, item); + + queue.Enqueue("x"); + + await Task.Delay(100); + + Assert.AreEqual("x", item); + Assert.AreEqual(0, queue.Count); + } + + [TestMethod] + public void Wait_For_Times() + { + var number = 0; + + var queue = new BlockingQueue(); + +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Run(async () => + { + while (true) + { + queue.Enqueue(1); + await Task.Delay(100); + } + }); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + + while (number < 50) + { + queue.Dequeue(); + Interlocked.Increment(ref number); + } + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj index 20cd0452..2baaa286 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj +++ b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj @@ -13,6 +13,7 @@ + diff --git a/Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs b/Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs new file mode 100644 index 00000000..670b8a1c --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs @@ -0,0 +1,38 @@ +using MQTTnet.Extensions.ManagedClient; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Server; + +namespace MQTTnet.Core.Tests +{ + [TestClass] + public class ManagedMqttClientTests + { + [TestMethod] + public async Task Drop_New_Messages_On_Full_Queue() + { + var factory = new MqttFactory(); + var managedClient = factory.CreateManagedMqttClient(); + + var clientOptions = new ManagedMqttClientOptionsBuilder() + .WithMaxPendingMessages(5) + .WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy.DropNewMessage); + + clientOptions.WithClientOptions(o => o.WithTcpServer("localhost")); + + await managedClient.StartAsync(clientOptions.Build()); + + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "1" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "2" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "3" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "4" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "5" }); + + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "6" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "7" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "8" }); + + Assert.AreEqual(5, managedClient.PendingApplicationMessagesCount); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index fc403771..b37dfdfa 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -340,6 +340,8 @@ namespace MQTTnet.Core.Tests } }); + await Task.Delay(100); + var retainedMessages = server.GetRetainedMessages(); Assert.AreEqual(ClientCount, retainedMessages.Count); diff --git a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs index 7b5ad77a..c405f5fe 100644 --- a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs +++ b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs @@ -45,7 +45,9 @@ namespace MQTTnet.Core.Tests stopwatch.Stop(); times.Add(stopwatch.Elapsed); stopwatch.Restart(); - } + } + + await server.StopAsync(); } } } From 95c30a28434b5f48a8a1563d46553e938500bb9a Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 29 Oct 2018 21:48:12 +0100 Subject: [PATCH 29/39] Update docs. --- Build/MQTTnet.AspNetCore.nuspec | 6 ++---- Build/MQTTnet.Extensions.ManagedClient.nuspec | 5 ++--- Build/MQTTnet.Extensions.Rpc.nuspec | 5 ++--- Build/MQTTnet.nuspec | 7 ++++--- 4 files changed, 10 insertions(+), 13 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 8809fac6..31ffecae 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -10,13 +10,11 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - * For more release notes please check the MQTTnet release notes. - * [Server] fixed cpu spike in case a client disconnectes (issue 421) - + For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). 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.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index e5d1184a..a24ab596 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -10,12 +10,11 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is an extension library which provides a managed MQTT client with additional features using MQTTnet. - * For more release notes please check the MQTTnet release notes. - + For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). 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 4075d654..038b6604 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -10,12 +10,11 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is an extension library which allows executing synchronous device calls including a response using MQTTnet. - * For more release notes please check the MQTTnet release notes. - + For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). 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 71f17dbe..423654b3 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,10 +10,11 @@ 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). - * [Server] Added new method which exposes all retained messages. -* [Server] Removed (wrong) setter from the server options interface. -* [ManagedClient] Added max pending messages count option. + * [ManagedClient] Added max pending messages count option. * [ManagedClient] Add pending messages overflow strategy option. +* [Server] Added new method which exposes all retained messages. +* [Server] Removed (wrong) setter from the server options interface. +* [Server] fixed cpu spike in case a client disconnectes (issue 421) 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 From 2bff21880624226f9713006d01643dbfe5d43651 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 29 Oct 2018 21:49:01 +0100 Subject: [PATCH 30/39] Update docs. --- Build/MQTTnet.nuspec | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 423654b3..d09d3e6f 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,7 +10,8 @@ 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). - * [ManagedClient] Added max pending messages count option. + * [Core] Updated nuget packages due to security issues. +* [ManagedClient] Added max pending messages count option. * [ManagedClient] Add pending messages overflow strategy option. * [Server] Added new method which exposes all retained messages. * [Server] Removed (wrong) setter from the server options interface. From a5caab62f83905a5ebbcac634db19d79768ed874 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 4 Nov 2018 11:39:12 +0100 Subject: [PATCH 31/39] fixed parallel writes --- .../MqttConnectionContext.cs | 18 ++++++++++++--- .../MqttConnectionContextTest.cs | 22 +++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 361eb6a1..e49b36f5 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -27,6 +27,8 @@ namespace MQTTnet.AspNetCore public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; + private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1); + public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) { if (Connection is TcpConnection tcp && !tcp.IsConnected) @@ -105,10 +107,20 @@ namespace MQTTnet.AspNetCore return null; } - public Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) + public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) { - var buffer = PacketSerializer.Serialize(packet); - return Connection.Transport.Output.WriteAsync(buffer.AsMemory(), cancellationToken).AsTask(); + var buffer = PacketSerializer.Serialize(packet).AsMemory(); + var output = Connection.Transport.Output; + + await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await output.WriteAsync(buffer, cancellationToken); + } + finally + { + _writerSemaphore.Release(); + } } public void Dispose() diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs index 7498015e..699aa65b 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs @@ -2,8 +2,10 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.AspNetCore.Tests.Mockups; using MQTTnet.Exceptions; +using MQTTnet.Packets; using MQTTnet.Serializer; using System; +using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -25,5 +27,25 @@ namespace MQTTnet.AspNetCore.Tests await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(TimeSpan.FromSeconds(1), CancellationToken.None)); } + + [TestMethod] + public async Task TestParallelWrites() + { + var serializer = new MqttPacketSerializer { }; + var pipe = new DuplexPipeMockup(); + var connection = new DefaultConnectionContext(); + connection.Transport = pipe; + var ctx = new MqttConnectionContext(serializer, connection); + + var tasks = Enumerable.Range(1, 10).Select(_ => Task.Run(async () => + { + for (int i = 0; i < 100; i++) + { + await ctx.SendPacketAsync(new MqttPublishPacket(), CancellationToken.None).ConfigureAwait(false); + } + })); + + await Task.WhenAll(tasks).ConfigureAwait(false); + } } } From ccadd4e41ac773e3c9602748c6b3f21883b0f1f0 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 4 Nov 2018 11:44:25 +0100 Subject: [PATCH 32/39] updated release notes --- Build/MQTTnet.AspNetCore.nuspec | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 31ffecae..59185142 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -10,7 +10,9 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). + For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). + * fixed concurrent sends with Aspnetcore.Connections.Abstractions based transport + 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 From 6fc9f12c34b6b301bcda6dd637901e8725e310f1 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 4 Nov 2018 13:18:42 +0100 Subject: [PATCH 33/39] Refactor task await. --- Source/MQTTnet.AspnetCore/MqttConnectionContext.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index e49b36f5..a5e567a7 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -62,7 +62,7 @@ namespace MQTTnet.AspNetCore } else { - readResult = await readTask; + readResult = await readTask.ConfigureAwait(false); } var buffer = readResult.Buffer; @@ -115,7 +115,7 @@ namespace MQTTnet.AspNetCore await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); try { - await output.WriteAsync(buffer, cancellationToken); + await output.WriteAsync(buffer, cancellationToken).ConfigureAwait(false); } finally { From 7181813f12d510a64fc76fb9b429858001513ac7 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 4 Nov 2018 20:33:44 +0100 Subject: [PATCH 34/39] Fix pending messages overflow strategy issue. --- Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 16d8e9b5..6dec29fe 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -116,7 +116,7 @@ namespace MQTTnet.Extensions.ManagedClient ManagedMqttApplicationMessage skippedMessage = null; lock (_messageQueue) { - if (_messageQueue.Count > _options.MaxPendingMessages) + if (_messageQueue.Count >= _options.MaxPendingMessages) { if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) { From 4e01c7f13d876e92352fd521b2abdd54a424c57c Mon Sep 17 00:00:00 2001 From: Jan Eggers Date: Mon, 5 Nov 2018 06:45:40 +0100 Subject: [PATCH 35/39] updated nuget spec according to csproj --- Build/MQTTnet.AspNetCore.nuspec | 8 ++++---- Build/MQTTnet.nuspec | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 59185142..f603215c 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -16,10 +16,10 @@ 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 d09d3e6f..1a95197d 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -24,7 +24,7 @@ - + @@ -33,7 +33,7 @@ - + From bddcbf69990233be9b6cf9f64a3123607e6e7f2a Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 5 Nov 2018 22:39:56 +0100 Subject: [PATCH 36/39] Refactor code. --- .../ManagedMqttClient.cs | 10 +++++----- .../ManagedMqttClientStorageManager.cs | 4 ++++ Source/MQTTnet/Internal/BlockingQueue.cs | 1 + 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 6dec29fe..271e6b93 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -113,7 +113,7 @@ namespace MQTTnet.Extensions.ManagedClient { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - ManagedMqttApplicationMessage skippedMessage = null; + ManagedMqttApplicationMessage removedMessage = null; lock (_messageQueue) { if (_messageQueue.Count >= _options.MaxPendingMessages) @@ -127,9 +127,9 @@ namespace MQTTnet.Extensions.ManagedClient if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage) { - skippedMessage = _messageQueue.RemoveFirst(); + removedMessage = _messageQueue.RemoveFirst(); _logger.Verbose("Removed oldest application message from internal queue because it is full."); - ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(skippedMessage)); + ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(removedMessage)); } } @@ -138,9 +138,9 @@ namespace MQTTnet.Extensions.ManagedClient if (_storageManager != null) { - if (skippedMessage != null) + if (removedMessage != null) { - await _storageManager.RemoveAsync(skippedMessage).ConfigureAwait(false); + await _storageManager.RemoveAsync(removedMessage).ConfigureAwait(false); } await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false); diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs index 0f4e33e3..821a7d92 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs @@ -28,6 +28,8 @@ namespace MQTTnet.Extensions.ManagedClient public async Task AddAsync(ManagedMqttApplicationMessage applicationMessage) { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { _messages.Add(applicationMessage); @@ -37,6 +39,8 @@ namespace MQTTnet.Extensions.ManagedClient public async Task RemoveAsync(ManagedMqttApplicationMessage applicationMessage) { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { var index = _messages.IndexOf(applicationMessage); diff --git a/Source/MQTTnet/Internal/BlockingQueue.cs b/Source/MQTTnet/Internal/BlockingQueue.cs index ae6cc985..d6c28f46 100644 --- a/Source/MQTTnet/Internal/BlockingQueue.cs +++ b/Source/MQTTnet/Internal/BlockingQueue.cs @@ -62,6 +62,7 @@ namespace MQTTnet.Internal { var item = _items.First; _items.RemoveFirst(); + return item.Value; } } From e9df698a1f28807d2dcf7063e4975e8a9c2d017a Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 5 Nov 2018 22:40:48 +0100 Subject: [PATCH 37/39] Update nuget versions. --- Build/MQTTnet.AspNetCore.nuspec | 2 +- Build/MQTTnet.Extensions.ManagedClient.nuspec | 2 +- Build/MQTTnet.Extensions.Rpc.nuspec | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index f603215c..21ef5d58 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -16,7 +16,7 @@ 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.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index a24ab596..572f73d3 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -14,7 +14,7 @@ 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 038b6604..29cec2b4 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -14,7 +14,7 @@ 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 - + From a1e258eb24cb4ded0e484ba5b5640daf69f66070 Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Wed, 14 Nov 2018 11:25:12 -0500 Subject: [PATCH 38/39] Check for pending disconnect in ThrowIfNotConnected This might be a little controversial, but it worked for us to correct a problem in which messages get stuck in the managed client storage queue (and are thrown out of the regular message queue without being published!) in the case of a failed connection. What we were seeing was that ManagedMqttClient.TryPublishQueuedMessage() was discarding a dequeued message without ever removing it from the storage queue because of an OperationCancelledException thrown by MqttClient.PublishAsync(). Tracing the code back, we found that when the connection is interrupted, after the timeout period MqttClient.InitiateDisconnect() would set the cancellation token, and the managed client would continue to try to publish, eventually calling through to MqttClient.SendAndReceiveAsync(), which would throw because the cancellation token is set. Looking back over the code, we saw that MqttClient.PublishAsync() has a call to ThrowIfNotConnected() at the top, which told us that the intent was to not allow this function to be called after a disconnect. But the disconnect was still pending, and the function wasn't behaving correctly in this state, so we reasoned that it's best to throw if the disconnect is pending. --- Source/MQTTnet/Client/MqttClient.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 05330b5d..bede2d34 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Client internal Task _keepAliveMessageSenderTask; private IMqttChannelAdapter _adapter; private bool _cleanDisconnectInitiated; - private int _disconnectGate; + private long _disconnectGate; public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger) { @@ -216,7 +216,7 @@ namespace MQTTnet.Client private void ThrowIfNotConnected() { - if (!IsConnected) throw new MqttCommunicationException("The client is not connected."); + if (!IsConnected || Interlocked.Read(ref _disconnectGate) == 1) throw new MqttCommunicationException("The client is not connected."); } private void ThrowIfConnected(string message) @@ -551,4 +551,4 @@ namespace MQTTnet.Client return Interlocked.CompareExchange(ref _disconnectGate, 1, 0) != 0; } } -} \ No newline at end of file +} From fcff3a2c8d003d75aa049e7807696c562d3c6f05 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 16 Nov 2018 22:12:29 +0100 Subject: [PATCH 39/39] Update docs. --- Build/MQTTnet.AspNetCore.nuspec | 4 +--- Build/MQTTnet.nuspec | 4 +++- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 21ef5d58..20489c78 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -10,9 +10,7 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). - * fixed concurrent sends with Aspnetcore.Connections.Abstractions based transport - + For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). 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 1a95197d..e3a8d290 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -11,11 +11,13 @@ false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). * [Core] Updated nuget packages due to security issues. +* [Client] Fixed wrong behavior of publish method when client is disconnecting (thanks to @PaulFake). * [ManagedClient] Added max pending messages count option. * [ManagedClient] Add pending messages overflow strategy option. * [Server] Added new method which exposes all retained messages. * [Server] Removed (wrong) setter from the server options interface. -* [Server] fixed cpu spike in case a client disconnectes (issue 421) +* [Server] fixed cpu spike in case a client disconnectes (issue 421). +* [Server] fixed concurrent sends with Aspnetcore.Connections.Abstractions based transport. 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