From 580a431fbf4b543f26c494ec320d9934f26061ee Mon Sep 17 00:00:00 2001 From: BP Date: Mon, 9 Apr 2018 10:59:34 +0200 Subject: [PATCH 1/6] Update Startup.cs --- Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index 9e5f90ae..591cc87d 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -5,6 +5,7 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.FileProviders; +using MQTTnet; using MQTTnet.AspNetCore; using MQTTnet.Server; From 9f4f37bb51c1e41e527ed84c0a9e9439cdacfb4a Mon Sep 17 00:00:00 2001 From: Christian Date: Sat, 28 Apr 2018 20:24:07 +0200 Subject: [PATCH 2/6] Update nugets --- Build/MQTTnet.AspNetCore.nuspec | 6 +++--- Build/MQTTnet.Extensions.Rpc.nuspec | 16 ++++++++-------- Build/MQTTnet.nuspec | 2 +- MQTTnet.sln | 1 + 4 files changed, 13 insertions(+), 12 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 664b09cd..4245077b 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -2,7 +2,7 @@ MQTTnet.AspNetCore - 2.7.4 + 2.7.5 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,13 +10,13 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - * Updated to MQTTnet 2.7.4. + * Updated to MQTTnet 2.7.5. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index ec09c1de..cb31c21b 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -2,7 +2,7 @@ MQTTnet.Extensions.Rpc - 2.6.0 + 2.7.5 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,32 +10,32 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a extension library which allows executing synchronous device calls including a response using MQTTnet. - * Initial version. + * Updated to MQTTnet 2.7.5. Copyright Christian Kratky 2016-2017 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 2b7bf1e6..4146cd38 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -2,7 +2,7 @@ MQTTnet - 2.7.4 + 2.7.5 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE diff --git a/MQTTnet.sln b/MQTTnet.sln index cece21c5..fab76088 100644 --- a/MQTTnet.sln +++ b/MQTTnet.sln @@ -18,6 +18,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1 appveyor.yml = appveyor.yml Build\build.ps1 = Build\build.ps1 Build\MQTTnet.AspNetCore.nuspec = Build\MQTTnet.AspNetCore.nuspec + Build\MQTTnet.Extensions.Rpc.nuspec = Build\MQTTnet.Extensions.Rpc.nuspec Build\MQTTnet.nuspec = Build\MQTTnet.nuspec EndProjectSection EndProject From 9fdcb08364316950867c98932fcfe3743a84abca Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 3 May 2018 20:29:02 +0200 Subject: [PATCH 3/6] Add benchmark app. --- MQTTnet.sln | 19 ++ Tests/MQTTnet.Benchmarks/App.config | 54 ++++++ .../MQTTnet.Benchmarks.csproj | 169 ++++++++++++++++++ .../MessageProcessingBenchmark.cs | 47 +++++ Tests/MQTTnet.Benchmarks/Program.cs | 14 ++ .../Properties/AssemblyInfo.cs | 36 ++++ Tests/MQTTnet.Benchmarks/packages.config | 54 ++++++ 7 files changed, 393 insertions(+) create mode 100644 Tests/MQTTnet.Benchmarks/App.config create mode 100644 Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj create mode 100644 Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs create mode 100644 Tests/MQTTnet.Benchmarks/Program.cs create mode 100644 Tests/MQTTnet.Benchmarks/Properties/AssemblyInfo.cs create mode 100644 Tests/MQTTnet.Benchmarks/packages.config diff --git a/MQTTnet.sln b/MQTTnet.sln index fab76088..a9bd6458 100644 --- a/MQTTnet.sln +++ b/MQTTnet.sln @@ -38,6 +38,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Extensions", "Extensions", EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.Rpc", "Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj", "{C444E9C8-95FA-430E-9126-274129DE16CD}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Benchmarks", "Tests\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{998D04DD-7CB0-45F5-A393-E2495C16399E}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -168,6 +170,22 @@ Global {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x64.Build.0 = Release|Any CPU {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x86.ActiveCfg = Release|Any CPU {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x86.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|Any CPU.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|ARM.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|ARM.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x64.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x64.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x86.ActiveCfg = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Debug|x86.Build.0 = Debug|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|Any CPU.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|Any CPU.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|ARM.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|ARM.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x64.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x64.Build.0 = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x86.ActiveCfg = Release|Any CPU + {998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -180,6 +198,7 @@ Global {C6FF8AEA-0855-41EC-A1F3-AC262225BAB9} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4} {F10C4060-F7EE-4A83-919F-FF723E72F94A} = {32A630A7-2598-41D7-B625-204CD906F5FB} {C444E9C8-95FA-430E-9126-274129DE16CD} = {12816BCC-AF9E-44A9-9AE5-C246AF2A0587} + {998D04DD-7CB0-45F5-A393-E2495C16399E} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894} diff --git a/Tests/MQTTnet.Benchmarks/App.config b/Tests/MQTTnet.Benchmarks/App.config new file mode 100644 index 00000000..5a41bc2c --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/App.config @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj new file mode 100644 index 00000000..49773119 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -0,0 +1,169 @@ + + + + + Debug + AnyCPU + {998D04DD-7CB0-45F5-A393-E2495C16399E} + Exe + MQTTnet.Benchmarks + MQTTnet.Benchmarks + v4.6.2 + 512 + true + + + AnyCPU + true + full + false + bin\Debug\ + DEBUG;TRACE + prompt + 4 + + + AnyCPU + pdbonly + true + bin\Release\ + TRACE + prompt + 4 + + + + ..\..\packages\BenchmarkDotNet.0.10.14\lib\net46\BenchmarkDotNet.dll + + + ..\..\packages\BenchmarkDotNet.Core.0.10.14\lib\net46\BenchmarkDotNet.Core.dll + + + ..\..\packages\BenchmarkDotNet.Toolchains.Roslyn.0.10.14\lib\net46\BenchmarkDotNet.Toolchains.Roslyn.dll + + + ..\..\packages\Microsoft.CodeAnalysis.Common.2.6.1\lib\netstandard1.3\Microsoft.CodeAnalysis.dll + + + ..\..\packages\Microsoft.CodeAnalysis.CSharp.2.6.1\lib\netstandard1.3\Microsoft.CodeAnalysis.CSharp.dll + + + ..\..\packages\Microsoft.DotNet.InternalAbstractions.1.0.0\lib\net451\Microsoft.DotNet.InternalAbstractions.dll + + + ..\..\packages\Microsoft.DotNet.PlatformAbstractions.1.1.1\lib\net451\Microsoft.DotNet.PlatformAbstractions.dll + + + ..\..\packages\Microsoft.Win32.Registry.4.3.0\lib\net46\Microsoft.Win32.Registry.dll + + + + ..\..\packages\System.AppContext.4.3.0\lib\net46\System.AppContext.dll + True + + + ..\..\packages\System.Collections.Immutable.1.3.1\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll + True + + + + ..\..\packages\System.Console.4.3.0\lib\net46\System.Console.dll + + + + ..\..\packages\System.Diagnostics.FileVersionInfo.4.3.0\lib\net46\System.Diagnostics.FileVersionInfo.dll + + + ..\..\packages\System.Diagnostics.StackTrace.4.3.0\lib\net46\System.Diagnostics.StackTrace.dll + + + ..\..\packages\System.IO.Compression.4.3.0\lib\net46\System.IO.Compression.dll + True + + + ..\..\packages\System.IO.FileSystem.4.3.0\lib\net46\System.IO.FileSystem.dll + + + ..\..\packages\System.IO.FileSystem.Primitives.4.3.0\lib\net46\System.IO.FileSystem.Primitives.dll + + + + + ..\..\packages\System.Reflection.4.3.0\lib\net462\System.Reflection.dll + + + ..\..\packages\System.Reflection.Metadata.1.4.2\lib\portable-net45+win8\System.Reflection.Metadata.dll + + + ..\..\packages\System.Runtime.4.3.0\lib\net462\System.Runtime.dll + + + ..\..\packages\System.Runtime.Extensions.4.3.0\lib\net462\System.Runtime.Extensions.dll + + + ..\..\packages\System.Runtime.InteropServices.4.3.0\lib\net462\System.Runtime.InteropServices.dll + + + ..\..\packages\System.Security.Cryptography.Algorithms.4.3.0\lib\net461\System.Security.Cryptography.Algorithms.dll + + + ..\..\packages\System.Security.Cryptography.Encoding.4.3.0\lib\net46\System.Security.Cryptography.Encoding.dll + + + ..\..\packages\System.Security.Cryptography.Primitives.4.3.0\lib\net46\System.Security.Cryptography.Primitives.dll + + + ..\..\packages\System.Security.Cryptography.X509Certificates.4.3.0\lib\net461\System.Security.Cryptography.X509Certificates.dll + + + ..\..\packages\System.Text.Encoding.CodePages.4.3.0\lib\net46\System.Text.Encoding.CodePages.dll + + + ..\..\packages\System.Threading.Tasks.Extensions.4.3.0\lib\portable-net45+win8+wp8+wpa81\System.Threading.Tasks.Extensions.dll + + + ..\..\packages\System.Threading.Thread.4.3.0\lib\net46\System.Threading.Thread.dll + + + ..\..\packages\System.ValueTuple.4.3.0\lib\netstandard1.0\System.ValueTuple.dll + + + + + + + + + ..\..\packages\System.Xml.ReaderWriter.4.3.0\lib\net46\System.Xml.ReaderWriter.dll + + + ..\..\packages\System.Xml.XmlDocument.4.3.0\lib\net46\System.Xml.XmlDocument.dll + + + ..\..\packages\System.Xml.XPath.4.3.0\lib\net46\System.Xml.XPath.dll + + + ..\..\packages\System.Xml.XPath.XDocument.4.3.0\lib\net46\System.Xml.XPath.XDocument.dll + + + + + + + + + + + + + + + + + + {3587E506-55A2-4EB3-99C7-DC01E42D25D2} + MQTTnet.NetStandard + + + + \ No newline at end of file diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs new file mode 100644 index 00000000..53b71fd6 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs @@ -0,0 +1,47 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Attributes.Columns; +using BenchmarkDotNet.Attributes.Exporters; +using BenchmarkDotNet.Attributes.Jobs; +using MQTTnet.Client; +using MQTTnet.Server; + +namespace MQTTnet.Benchmarks +{ + [ClrJob] + [RPlotExporter, RankColumn] + public class MessageProcessingBenchmark + { + private IMqttServer _mqttServer; + private IMqttClient _mqttClient; + private MqttApplicationMessage _message; + + [GlobalSetup] + public void Setup() + { + var factory = new MqttFactory(); + _mqttServer = factory.CreateMqttServer(); + _mqttClient = factory.CreateMqttClient(); + + var serverOptions = new MqttServerOptionsBuilder().Build(); + _mqttServer.StartAsync(serverOptions).GetAwaiter().GetResult(); + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost").Build(); + + _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult(); + + _message = new MqttApplicationMessageBuilder() + .WithTopic("A") + .Build(); + } + + [Benchmark] + public void Send_10000_Messages() + { + for (var i = 0; i < 10000; i++) + { + _mqttClient.PublishAsync(_message).GetAwaiter().GetResult(); + } + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs new file mode 100644 index 00000000..965ddb55 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/Program.cs @@ -0,0 +1,14 @@ +using System; +using BenchmarkDotNet.Running; + +namespace MQTTnet.Benchmarks +{ + public static class Program + { + public static void Main(string[] args) + { + var summary = BenchmarkRunner.Run(); + Console.ReadLine(); + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Properties/AssemblyInfo.cs b/Tests/MQTTnet.Benchmarks/Properties/AssemblyInfo.cs new file mode 100644 index 00000000..4df83f6e --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/Properties/AssemblyInfo.cs @@ -0,0 +1,36 @@ +using System.Reflection; +using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; + +// General Information about an assembly is controlled through the following +// set of attributes. Change these attribute values to modify the information +// associated with an assembly. +[assembly: AssemblyTitle("MQTTnet.Benchmarks")] +[assembly: AssemblyDescription("")] +[assembly: AssemblyConfiguration("")] +[assembly: AssemblyCompany("")] +[assembly: AssemblyProduct("MQTTnet.Benchmarks")] +[assembly: AssemblyCopyright("Copyright © 2018")] +[assembly: AssemblyTrademark("")] +[assembly: AssemblyCulture("")] + +// Setting ComVisible to false makes the types in this assembly not visible +// to COM components. If you need to access a type in this assembly from +// COM, set the ComVisible attribute to true on that type. +[assembly: ComVisible(false)] + +// The following GUID is for the ID of the typelib if this project is exposed to COM +[assembly: Guid("998d04dd-7cb0-45f5-a393-e2495c16399e")] + +// Version information for an assembly consists of the following four values: +// +// Major Version +// Minor Version +// Build Number +// Revision +// +// You can specify all the values or you can default the Build and Revision Numbers +// by using the '*' as shown below: +// [assembly: AssemblyVersion("1.0.*")] +[assembly: AssemblyVersion("1.0.0.0")] +[assembly: AssemblyFileVersion("1.0.0.0")] diff --git a/Tests/MQTTnet.Benchmarks/packages.config b/Tests/MQTTnet.Benchmarks/packages.config new file mode 100644 index 00000000..bb930145 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/packages.config @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file From d9501de252da2aa52e822c0892041a81f08af016 Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 3 May 2018 20:29:53 +0200 Subject: [PATCH 4/6] Treat a connection handover as a clean disconnect. --- .../Server/MqttClientDisconnectType.cs | 8 +++++ .../Server/MqttClientPendingMessagesQueue.cs | 2 +- .../Server/MqttClientSession.cs | 33 ++++++++++--------- .../Server/MqttClientSessionsManager.cs | 6 ++-- 4 files changed, 30 insertions(+), 19 deletions(-) create mode 100644 Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs new file mode 100644 index 00000000..19e9da8d --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs @@ -0,0 +1,8 @@ +namespace MQTTnet.Server +{ + public enum MqttClientDisconnectType + { + Clean, + NotClean + } +} diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs index 99493874..1602a4d1 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs @@ -128,7 +128,7 @@ namespace MQTTnet.Server if (!cancellationToken.IsCancellationRequested) { - await _clientSession.StopAsync().ConfigureAwait(false); + await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs index 8514da99..f86f72a1 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs @@ -90,7 +90,7 @@ namespace MQTTnet.Server return _wasCleanDisconnect; } - public async Task StopAsync(bool wasCleanDisconnect = false) + public async Task StopAsync(MqttClientDisconnectType type) { try { @@ -99,7 +99,7 @@ namespace MQTTnet.Server return; } - _wasCleanDisconnect = wasCleanDisconnect; + _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; _cancellationTokenSource?.Cancel(false); @@ -118,7 +118,7 @@ namespace MQTTnet.Server var willMessage = _willMessage; _willMessage = null; // clear willmessage so it is send just once - if (willMessage != null && !wasCleanDisconnect) + if (willMessage != null && !_wasCleanDisconnect) { await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false); } @@ -181,7 +181,7 @@ namespace MQTTnet.Server private Task StopDueToKeepAliveTimeoutAsync() { _logger.Info("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId); - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) @@ -198,15 +198,18 @@ namespace MQTTnet.Server catch (OperationCanceledException) { } - catch (MqttCommunicationException exception) - { - _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); - await StopAsync().ConfigureAwait(false); - } catch (Exception exception) { - _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); - await StopAsync().ConfigureAwait(false); + if (exception is MqttCommunicationException) + { + _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); + } + else + { + _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); + } + + await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } } @@ -255,16 +258,16 @@ namespace MQTTnet.Server if (packet is MqttDisconnectPacket) { - return StopAsync(true); + return StopAsync(MqttClientDisconnectType.Clean); } if (packet is MqttConnectPacket) { - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } _logger.Warning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) @@ -274,7 +277,7 @@ namespace MQTTnet.Server if (subscribeResult.CloseConnection) { - await StopAsync().ConfigureAwait(false); + await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index 8dd9d062..5f6cf15e 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -124,7 +124,7 @@ namespace MQTTnet.Server { foreach (var session in _sessions) { - await session.Value.StopAsync().ConfigureAwait(false); + await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } _sessions.Clear(); @@ -162,7 +162,7 @@ namespace MQTTnet.Server var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); if (interceptorContext.CloseConnection) { - await senderClientSession.StopAsync().ConfigureAwait(false); + await senderClientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) @@ -283,7 +283,7 @@ namespace MQTTnet.Server { _sessions.Remove(connectPacket.ClientId); - await clientSession.StopAsync().ConfigureAwait(false); + await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false); clientSession.Dispose(); clientSession = null; From aaed2ed790561a2d079590643ad26b69d809a4a3 Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 3 May 2018 20:30:40 +0200 Subject: [PATCH 5/6] Process client messages in worker thread (extend options) --- .../ApplicationBuilderExtensions.cs | 10 +- .../Client/IMqttClientOptions.cs | 11 +- .../MQTTnet.NetStandard/Client/MqttClient.cs | 298 ++++++++++-------- .../Client/MqttClientOptions.cs | 17 +- .../Client/MqttClientOptionsBuilder.cs | 7 + ...eceivedApplicationMessageProcessingMode.cs | 8 + 6 files changed, 192 insertions(+), 159 deletions(-) create mode 100644 Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs diff --git a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs index 5a25712d..87b2038b 100644 --- a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs @@ -17,12 +17,12 @@ namespace MQTTnet.AspNetCore { string subprotocol = null; - if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues) - && requestedSubProtocolValues.Count > 0 - && requestedSubProtocolValues.Any(v => v.ToLower() == "mqtt") - ) + if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues)) { - subprotocol = "mqtt"; + // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc. + subprotocol = requestedSubProtocolValues + .OrderByDescending(p => p.Length) + .FirstOrDefault(p => p.ToLower().StartsWith("mqtt")); } var adapter = app.ApplicationServices.GetRequiredService(); diff --git a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs index 647f688b..aa873af0 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs @@ -6,17 +6,16 @@ namespace MQTTnet.Client public interface IMqttClientOptions { string ClientId { get; } - - IMqttClientCredentials Credentials { get; } bool CleanSession { get; } - MqttApplicationMessage WillMessage { get; } + IMqttClientCredentials Credentials { get; } + MqttProtocolVersion ProtocolVersion { get; } + IMqttClientChannelOptions ChannelOptions { get; } TimeSpan CommunicationTimeout { get; } TimeSpan KeepAlivePeriod { get; } TimeSpan? KeepAliveSendInterval { get; } + MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; } - MqttProtocolVersion ProtocolVersion { get; } - - IMqttClientChannelOptions ChannelOptions { get; } + MqttApplicationMessage WillMessage { get; } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs index 90790d5a..ec863563 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs @@ -23,7 +23,6 @@ namespace MQTTnet.Client private readonly IMqttNetLogger _logger; private IMqttClientOptions _options; - private bool _isReceivingPackets; private CancellationTokenSource _cancellationTokenSource; private Task _packetReceiverTask; private Task _keepAliveMessageSenderTask; @@ -63,16 +62,16 @@ namespace MQTTnet.Client await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false); _logger.Verbose("Connection with server established."); - await StartReceivingPacketsAsync().ConfigureAwait(false); + StartReceivingPackets(_cancellationTokenSource.Token); - var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false); + var connectResponse = await AuthenticateAsync(options.WillMessage, _cancellationTokenSource.Token).ConfigureAwait(false); _logger.Verbose("MQTT connection with server established."); _sendTracker.Restart(); if (_options.KeepAlivePeriod != TimeSpan.Zero) { - StartSendingKeepAliveMessages(); + StartSendingKeepAliveMessages(_cancellationTokenSource.Token); } IsConnected = true; @@ -101,7 +100,7 @@ namespace MQTTnet.Client { if (!_cancellationTokenSource.IsCancellationRequested) { - await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false); + await SendAsync(new MqttDisconnectPacket(), _cancellationTokenSource.Token).ConfigureAwait(false); } } finally @@ -122,7 +121,7 @@ namespace MQTTnet.Client TopicFilters = topicFilters.ToList() }; - var response = await SendAndReceiveAsync(subscribePacket).ConfigureAwait(false); + var response = await SendAndReceiveAsync(subscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false); if (response.SubscribeReturnCodes.Count != subscribePacket.TopicFilters.Count) { @@ -144,7 +143,7 @@ namespace MQTTnet.Client TopicFilters = topicFilters.ToList() }; - await SendAndReceiveAsync(unsubscribePacket).ConfigureAwait(false); + await SendAndReceiveAsync(unsubscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false); } public async Task PublishAsync(IEnumerable applicationMessages) @@ -161,7 +160,7 @@ namespace MQTTnet.Client case MqttQualityOfServiceLevel.AtMostOnce: { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await SendAsync(qosGroup.Cast().ToArray()).ConfigureAwait(false); + await SendAsync(qosGroup, _cancellationTokenSource.Token).ConfigureAwait(false); break; } case MqttQualityOfServiceLevel.AtLeastOnce: @@ -169,7 +168,7 @@ namespace MQTTnet.Client foreach (var publishPacket in qosGroup) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); - await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); + await SendAndReceiveAsync(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false); } break; @@ -180,13 +179,13 @@ namespace MQTTnet.Client { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); - var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); + var pubRecPacket = await SendAndReceiveAsync(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false); var pubRelPacket = new MqttPubRelPacket { PacketIdentifier = pubRecPacket.PacketIdentifier }; - await SendAndReceiveAsync(pubRelPacket).ConfigureAwait(false); + await SendAndReceiveAsync(pubRelPacket, _cancellationTokenSource.Token).ConfigureAwait(false); } break; @@ -207,7 +206,7 @@ namespace MQTTnet.Client _adapter?.Dispose(); } - private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage) + private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) { var connectPacket = new MqttConnectPacket { @@ -219,7 +218,7 @@ namespace MQTTnet.Client WillMessage = willApplicationMessage }; - var response = await SendAndReceiveAsync(connectPacket).ConfigureAwait(false); + var response = await SendAndReceiveAsync(connectPacket, _cancellationTokenSource.Token).ConfigureAwait(false); if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { throw new MqttConnectingFailedException(response.ConnectReturnCode); @@ -264,21 +263,19 @@ namespace MQTTnet.Client try { - if (_packetReceiverTask != null && _packetReceiverTask != sender) - { - _packetReceiverTask.Wait(); - } + await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); + await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false); if (_keepAliveMessageSenderTask != null && _keepAliveMessageSenderTask != sender) { - _keepAliveMessageSenderTask.Wait(); + await _keepAliveMessageSenderTask.ConfigureAwait(false); } if (_adapter != null) { await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false); } - + _logger.Verbose("Disconnected from adapter."); } catch (Exception adapterException) @@ -297,101 +294,19 @@ namespace MQTTnet.Client } } - private async Task ProcessReceivedPacketAsync(MqttBasePacket packet) - { - try - { - if (packet is MqttPublishPacket publishPacket) - { - await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false); - return; - } - - if (packet is MqttPingReqPacket) - { - await SendAsync(new MqttPingRespPacket()).ConfigureAwait(false); - return; - } - - if (packet is MqttDisconnectPacket) - { - await DisconnectAsync().ConfigureAwait(false); - return; - } - - if (packet is MqttPubRelPacket pubRelPacket) - { - await ProcessReceivedPubRelPacket(pubRelPacket).ConfigureAwait(false); - return; - } - - _packetDispatcher.Dispatch(packet); - } - catch (Exception exception) - { - _logger.Error(exception, "Unhandled exception while processing received packet."); - } - } - - private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) - { - try - { - var applicationMessage = publishPacket.ToApplicationMessage(); - ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); - } - catch (Exception exception) - { - _logger.Error(exception, "Unhandled exception while handling application message."); - } - } - - private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket) - { - if (_cancellationTokenSource.IsCancellationRequested) - { - return Task.FromResult(0); - } - - if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) - { - FireApplicationMessageReceivedEvent(publishPacket); - return Task.FromResult(0); - } - - if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) - { - FireApplicationMessageReceivedEvent(publishPacket); - return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); - } - - if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) - { - // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] - FireApplicationMessageReceivedEvent(publishPacket); - return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); - } - - throw new MqttCommunicationException("Received a not supported QoS level."); - } - - private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket) - { - var response = new MqttPubCompPacket - { - PacketIdentifier = pubRelPacket.PacketIdentifier - }; - - return SendAsync(response); - } - - private Task SendAsync(params MqttBasePacket[] packets) + private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken) { _sendTracker.Restart(); - return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packets); + return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, new[] { packet }); } - private async Task SendAndReceiveAsync(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket + private Task SendAsync(IEnumerable packets, CancellationToken cancellationToken) + { + _sendTracker.Restart(); + return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, packets); + } + + private async Task SendAndReceiveAsync(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket { ushort? identifier = null; if (requestPacket is IMqttPacketWithIdentifier requestPacketWithIdentifier) @@ -400,18 +315,18 @@ namespace MQTTnet.Client } var packetAwaiter = _packetDispatcher.WaitForPacketAsync(typeof(TResponsePacket), identifier, _options.CommunicationTimeout); - await SendAsync(requestPacket).ConfigureAwait(false); + await SendAsync(requestPacket, cancellationToken).ConfigureAwait(false); return (TResponsePacket)await packetAwaiter.ConfigureAwait(false); } - private async Task SendKeepAliveMessagesAsync() + private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken) { _logger.Verbose("Start sending keep alive packets."); try { - while (!_cancellationTokenSource.Token.IsCancellationRequested) + while (!cancellationToken.IsCancellationRequested) { var keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75); if (_options.KeepAliveSendInterval.HasValue) @@ -421,10 +336,10 @@ namespace MQTTnet.Client if (_sendTracker.Elapsed > keepAliveSendInterval) { - await SendAndReceiveAsync(new MqttPingReqPacket()).ConfigureAwait(false); + await SendAndReceiveAsync(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false); } - await Task.Delay(keepAliveSendInterval, _cancellationTokenSource.Token).ConfigureAwait(false); + await Task.Delay(keepAliveSendInterval, cancellationToken).ConfigureAwait(false); } } catch (Exception exception) @@ -440,7 +355,7 @@ namespace MQTTnet.Client { _logger.Error(exception, "Unhandled exception while sending/receiving keep alive packets."); } - + await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false); } finally @@ -449,24 +364,34 @@ namespace MQTTnet.Client } } - private async Task ReceivePacketsAsync() + private async Task ReceivePacketsAsync(CancellationToken cancellationToken) { _logger.Verbose("Start receiving packets."); try { - while (!_cancellationTokenSource.Token.IsCancellationRequested) + while (!cancellationToken.IsCancellationRequested) { - _isReceivingPackets = true; + var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); - var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false); - - if (_cancellationTokenSource.Token.IsCancellationRequested) + if (cancellationToken.IsCancellationRequested) { return; } - StartProcessReceivedPacket(packet); + if (packet == null) + { + continue; + } + + if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.SingleThread) + { + await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); + } + else if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.DedicatedThread) + { + StartProcessReceivedPacketAsync(packet, cancellationToken); + } } } catch (Exception exception) @@ -491,26 +416,125 @@ namespace MQTTnet.Client } } - private void StartProcessReceivedPacket(MqttBasePacket packet) + private async Task ProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) { - Task.Run(() => ProcessReceivedPacketAsync(packet), _cancellationTokenSource.Token); - } - - private async Task StartReceivingPacketsAsync() - { - _isReceivingPackets = false; - - _packetReceiverTask = Task.Run(ReceivePacketsAsync, _cancellationTokenSource.Token); - - while (!_isReceivingPackets && !_cancellationTokenSource.Token.IsCancellationRequested) + try { - await Task.Delay(TimeSpan.FromMilliseconds(100), _cancellationTokenSource.Token).ConfigureAwait(false); + if (packet is MqttPublishPacket publishPacket) + { + await ProcessReceivedPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); + return; + } + + if (packet is MqttPingReqPacket) + { + await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false); + return; + } + + if (packet is MqttDisconnectPacket) + { + await DisconnectAsync().ConfigureAwait(false); + return; + } + + if (packet is MqttPubRelPacket pubRelPacket) + { + await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false); + return; + } + + _packetDispatcher.Dispatch(packet); + } + catch (Exception exception) + { + _logger.Error(exception, "Unhandled exception while processing received packet."); } } - private void StartSendingKeepAliveMessages() + private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { - _keepAliveMessageSenderTask = Task.Run(SendKeepAliveMessagesAsync, _cancellationTokenSource.Token); + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) + { + FireApplicationMessageReceivedEvent(publishPacket); + return Task.FromResult(0); + } + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) + { + FireApplicationMessageReceivedEvent(publishPacket); + return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken); + } + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) + { + // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] + FireApplicationMessageReceivedEvent(publishPacket); + return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken); + } + + throw new MqttCommunicationException("Received a not supported QoS level."); + } + + private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) + { + var response = new MqttPubCompPacket + { + PacketIdentifier = pubRelPacket.PacketIdentifier + }; + + return SendAsync(response, cancellationToken); + } + + private void StartReceivingPackets(CancellationToken cancellationToken) + { + _packetReceiverTask = Task.Run(() => ReceivePacketsAsync(cancellationToken), cancellationToken); + } + + private void StartSendingKeepAliveMessages(CancellationToken cancellationToken) + { + _keepAliveMessageSenderTask = Task.Run(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken); + } + + private void StartProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) + { +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Run(() => ProcessReceivedPacketAsync(packet, cancellationToken), cancellationToken); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + } + + private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) + { + try + { + var applicationMessage = publishPacket.ToApplicationMessage(); + ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); + } + catch (Exception exception) + { + _logger.Error(exception, "Unhandled exception while handling application message."); + } + } + + private static async Task WaitForTaskAsync(Task task, Task sender) + { + if (task == sender) + { + return; + } + + if (task.IsCanceled || task.IsCompleted || task.IsFaulted) + { + return; + } + + try + { + await task.ConfigureAwait(false); + } + catch (TaskCanceledException) + { + } } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs index 9640a474..ae0517a6 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs @@ -5,22 +5,17 @@ namespace MQTTnet.Client { public class MqttClientOptions : IMqttClientOptions { - public MqttApplicationMessage WillMessage { get; set; } - public string ClientId { get; set; } = Guid.NewGuid().ToString("N"); - public bool CleanSession { get; set; } = true; - public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials(); - - public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15); - - public TimeSpan? KeepAliveSendInterval { get; set; } + public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; + public IMqttClientChannelOptions ChannelOptions { get; set; } public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10); + public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15); + public TimeSpan? KeepAliveSendInterval { get; set; } + public MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; set; } = MqttReceivedApplicationMessageProcessingMode.SingleThread; - public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; - - public IMqttClientChannelOptions ChannelOptions { get; set; } + public MqttApplicationMessage WillMessage { get; set; } } } diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs index ad766c3f..bc6aa9f6 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs @@ -108,6 +108,13 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithReceivedApplicationMessageProcessingMode( + MqttReceivedApplicationMessageProcessingMode mode) + { + _options.ReceivedApplicationMessageProcessingMode = mode; + return this; + } + public IMqttClientOptions Build() { if (_tlsOptions != null) diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs new file mode 100644 index 00000000..651abe5b --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs @@ -0,0 +1,8 @@ +namespace MQTTnet.Client +{ + public enum MqttReceivedApplicationMessageProcessingMode + { + SingleThread, + DedicatedThread + } +} From ff45be1f7158e8ac5b93854a8559cbc2679f33e2 Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 3 May 2018 20:30:44 +0200 Subject: [PATCH 6/6] Update docs --- Build/MQTTnet.nuspec | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 4146cd38..764a36c5 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -2,7 +2,7 @@ MQTTnet - 2.7.5 + 2.8.0 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,13 +10,9 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). - * [Client] Fixed a deadlock while the client disconnects. -* [Client] Fixed broken support for protocol version 3.1.0. -* [Server] The _MqttTcpServerAdapter_ is now added to the ASP.NET services. -* [Server] _MqttServerAdapter_ is renamed to _MqttTcpServerAdapter_ (BREAKING CHANGE!). -* [Server] The server no longer sends the will message of a client if the disconnect was clean (via _Disconnect_ packet). -* [Server] The application message interceptor now allows closing the connection. -* [Server] Added a new flag for the _ClientDisconnected_ event which contains a value indicating whether the disconnect was clean (via _Disconnect_ packet). + * [Client] Received messages are now processed in the worker thread by default. Added a new setting for switching back to dedicated threads. +* [Server] Added support for other WebSocket sub protocol formats like mqttv-3.1.1 (thanks to @israellot). +* [Server] The takeover of an existing client sessions is now treated as a _clean_ disconnect of the previous client. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin