Compare commits

...
47 Commits
Author SHA1 Message Date
Christian Kratky 763f0e7934 Update nuspec. 2020-05-14 21:01:39 +02:00
Christian Kratky 9d381c2cf9 Update nuspec. 2020-05-14 00:40:14 +02:00
Christian Kratky 17ceb20169 Add dummy method to allow safe downgrade. 2020-05-14 00:18:50 +02:00
Christian Kratky 683bb43486 Fix performance issue in logger. 2020-05-14 00:05:52 +02:00
Christian Kratky d25b51740b Merge remote-tracking branch 'origin/master' 2020-05-12 21:55:17 +02:00
Christian Kratky a58ab607ad Improve logger performance. 2020-05-12 21:55:15 +02:00
Christian 0a7d450f3b Merge pull request #920 from mseyock/change-icon
Remove shadow from icon
2020-05-08 16:37:31 +02:00
ms 999fa7a468 Remove shadow from icon 2020-05-08 15:21:50 +02:00
Christian Kratky f0a1f9ca93 Add support for user properties in client connect. 2020-05-04 23:15:38 +02:00
Christian Kratky 89e9cb224f Update nuspec. 2020-05-02 21:20:22 +02:00
Christian Kratky fe48c9c837 Update nuspecs. 2020-05-01 23:39:17 +02:00
Christian Kratky 2a421d1d42 Refactor code. 2020-05-01 21:47:25 +02:00
Christian Kratky 48d0344a40 Fix WebSocket4Net extension nuspec. 2020-04-30 23:15:28 +02:00
Christian Kratky db5ed594f7 Merge remote-tracking branch 'origin/master' 2020-04-30 23:02:09 +02:00
Christian Kratky f85bc9b956 Dispose socket when TCP connect failed. 2020-04-30 23:02:05 +02:00
Christian eec46dd9bc Update LICENSE 2020-04-30 22:07:50 +02:00
Christian 9c9e2293c8 Update README.md 2020-04-29 18:04:17 +02:00
Christian 856861e1b1 Merge pull request #909 from JanEggers/nuget
removed unused dependecies
2020-04-27 21:24:24 +02:00
JanEggers 4b481b4833 removed unused dependecies 2020-04-27 17:09:54 +02:00
Christian c05f11a495 Update README.md 2020-04-25 14:44:07 +02:00
Christian Kratky e798c91ff3 Fix exception handling when using invalid host names upon connect. 2020-04-24 19:29:03 +02:00
Christian Kratky 5df4fc3d5d Update docs. 2020-04-22 21:10:55 +02:00
Christian Kratky c6236c0865 Remove nuget config. 2020-04-22 21:09:48 +02:00
Christian Kratky 85c0cb3bcc Remove nuget config because it breaks the CI. 2020-04-22 12:32:57 +02:00
Christian Kratky 9810e52bcb Fix build. 2020-04-22 12:30:11 +02:00
Christian Kratky 948a5eaae3 Merge remote-tracking branch 'origin/master' 2020-04-22 12:05:11 +02:00
Christian Kratky 31df8247fa Try to fix CI build. 2020-04-22 12:05:05 +02:00
Christian 229418100c Merge pull request #904 from JanEggers/nuget
fixed netcoreapp3.1 build #903
2020-04-22 12:04:02 +02:00
JanEggers 93d8679bbc fixed netcoreapp3.1 build 2020-04-21 21:19:49 +02:00
Christian 6cec126862 Update README.md 2020-04-19 01:11:15 +02:00
Christian Kratky 9749f790ea Try to fix CI builds. 2020-04-18 20:35:39 +02:00
Christian Kratky fbbc638c75 Try to fix CI builds. 2020-04-18 20:28:46 +02:00
Christian Kratky bf679adb6b Try to fix CI builds. 2020-04-18 19:43:28 +02:00
Christian Kratky cf7e59a774 Improve unit tests. 2020-04-18 16:55:37 +02:00
Christian e9724c0202 Merge pull request #882 from xljiulang/master
Use IConnectionEndPointFeature to get IMqttChannelAdapter.Endpoint in aspnetcore31
2020-04-18 13:49:45 +02:00
Christian Kratky 7931af7487 Add support for certificate validation callback for Web Socket connections. 2020-04-18 13:46:20 +02:00
Christian Kratky bb5f023b74 Update docs. 2020-04-18 01:38:53 +02:00
Christian Kratky c2ac4fffaf Expose PING method for clients. 2020-04-18 01:38:48 +02:00
Christian Kratky 4b78710705 Refactor topic filter naming. 2020-04-18 01:38:20 +02:00
Christian Kratky 2e69f8be78 Merge remote-tracking branch 'origin/master' 2020-04-16 12:48:31 +02:00
Christian Kratky 2d6c44e45b MQTTnet.Server: Fix version detection. 2020-04-16 12:48:26 +02:00
Christian f001d7aecc Merge pull request #898 from xgstation/master
Fix hanging issue under UWP.
2020-04-14 22:21:31 +02:00
JanEggers d49a4384c0 use the RemoteEndpoint property from the ConnectionContext and added test 2020-04-14 19:16:32 +02:00
Gen 4e6658e014 Fix typo. 2020-04-14 09:36:02 -07:00
Gen 2b2b30912c Configure context. 2020-04-14 07:37:23 -07:00
Gen 79d95c8516 Fix hanging issue under UWP. 2020-04-13 03:04:44 -07:00
陈国伟 c10145f743 Use IConnectionEndPointFeature to get IMqttChannelAdapter.Endpoint in aspnetcore31 2020-03-26 12:32:09 +08:00
131 changed files with 1188 additions and 883 deletions
+5
View File
@@ -0,0 +1,5 @@
[*.cs]
# CA1031: Do not catch general exception types
dotnet_diagnostic.CA1031.severity = none
+13 -8
View File
@@ -3,24 +3,26 @@
<metadata>
<id>MQTTnet.AspNetCore</id>
<version>0.0.0</version>
<authors>Christian Kratky</authors>
<authors>The contributors of MQTTnet</authors>
<owners>Christian Kratky</owners>
<license type="file">LICENSE</license>
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<icon>images\Logo_128x128.png</icon>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<requireLicenseAcceptance>true</requireLicenseAcceptance>
<description>This is a support library to integrate MQTTnet into AspNetCore.</description>
<releaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2020</copyright>
<tags>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</tags>
<dependencies>
<dependency id="MQTTnet" version="$nugetVersion" />
<dependency id="Microsoft.AspNetCore.Connections.Abstractions" version="3.1.3" />
<dependency id="Microsoft.AspNetCore.Http.Connections" version="1.1.0" />
<dependency id="Microsoft.AspNetCore.WebSockets" version="2.2.1" />
<dependency id="Microsoft.Extensions.Hosting.Abstractions" version="3.1.3" />
<group targetFramework="netstandard2.0">
<dependency id="MQTTnet" version="$nugetVersion" />
<dependency id="Microsoft.AspNetCore.Http.Connections" version="1.0.0" />
</group>
<group targetFramework="netcoreapp3.1">
<dependency id="MQTTnet" version="$nugetVersion" />
</group>
</dependencies>
</metadata>
@@ -33,5 +35,8 @@
<!-- .NET Standard 2.0 -->
<file src="..\Source\MQTTnet.AspNetCore\bin\Release\netstandard2.0\MQTTnet.AspNetCore.*" target="lib\netstandard2.0\"/>
<!-- .NET Core App 3.1 -->
<file src="..\Source\MQTTnet.AspNetCore\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.*" target="lib\netcoreapp3.1\"/>
</files>
</package>
@@ -3,13 +3,13 @@
<metadata>
<id>MQTTnet.Extensions.ManagedClient</id>
<version>0.0.0</version>
<authors>Christian Kratky</authors>
<authors>The contributors of MQTTnet</authors>
<owners>Christian Kratky</owners>
<license type="file">LICENSE</license>
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<icon>images\Logo_128x128.png</icon>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<requireLicenseAcceptance>true</requireLicenseAcceptance>
<description>This is an extension library which provides a managed MQTT client with additional features using MQTTnet.</description>
<releaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2020</copyright>
+2 -2
View File
@@ -3,13 +3,13 @@
<metadata>
<id>MQTTnet.Extensions.Rpc</id>
<version>0.0.0</version>
<authors>Christian Kratky</authors>
<authors>The contributors of MQTTnet</authors>
<owners>Christian Kratky</owners>
<license type="file">LICENSE</license>
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<icon>images\Logo_128x128.png</icon>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<requireLicenseAcceptance>true</requireLicenseAcceptance>
<description>This is an extension library which allows executing synchronous device calls including a response using MQTTnet.</description>
<releaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2020</copyright>
+18 -3
View File
@@ -3,19 +3,34 @@
<metadata>
<id>MQTTnet.Extensions.WebSocket4Net</id>
<version>0.0.0</version>
<authors>Christian Kratky</authors>
<authors>The contributors of MQTTnet</authors>
<owners>Christian Kratky</owners>
<license type="file">LICENSE</license>
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<icon>images\Logo_128x128.png</icon>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<requireLicenseAcceptance>true</requireLicenseAcceptance>
<description>This is an extension library which allows using _WebSocket4Net_ as transport for MQTTnet clients.</description>
<releaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2020</copyright>
<tags>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</tags>
<dependencies>
<dependency id="MQTTnet" version="$nugetVersion" />
<group targetFramework="netstandard1.3">
<dependency id="MQTTnet" version="$nugetVersion" />
<dependency id="WebSocket4Net" version="0.15.2" />
</group>
<group targetFramework="netstandard2.0">
<dependency id="MQTTnet" version="$nugetVersion" />
<dependency id="WebSocket4Net" version="0.15.2" />
</group>
<group targetFramework="netstandard2.1">
<dependency id="MQTTnet" version="$nugetVersion" />
<dependency id="WebSocket4Net" version="0.15.2" />
</group>
<group targetFramework="uap10.0">
<dependency id="MQTTnet" version="$nugetVersion" />
<dependency id="WebSocket4Net" version="0.15.2" />
</group>
</dependencies>
</metadata>
-46
View File
@@ -1,46 +0,0 @@
<?xml version="1.0"?>
<package >
<metadata>
<id>MQTTnet.NETStandard</id>
<version>0.0.0</version>
<authors>Christian Kratky</authors>
<owners>Christian Kratky</owners>
<license type="file">LICENSE</license>
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<icon>images\Logo_128x128.png</icon>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<description>This package contains the .NET Standard version of MQTTnet only.</description>
<releaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2020</copyright>
<tags>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</tags>
<dependencies>
<group targetFramework="netstandard1.3">
<dependency id="NETStandard.Library" version="1.3.0" />
<dependency id="System.Net.Security" version="4.3.2" />
<dependency id="System.Net.WebSockets" version="4.3.0" />
<dependency id="System.Net.WebSockets.Client" version="4.3.2" />
</group>
<group targetFramework="netstandard2.0">
<dependency id="NETStandard.Library" version="2.0.0" />
<dependency id="System.Net.Security" version="4.3.2" />
<dependency id="System.Net.WebSockets" version="4.3.0" />
<dependency id="System.Net.WebSockets.Client" version="4.3.2" />
</group>
</dependencies>
</metadata>
<files>
<!-- License -->
<file src="..\LICENSE" />
<!-- Images -->
<file src="..\Images\Logo_128x128.png" target="images\" />
<!-- .NET Standard 1.3 -->
<file src="..\Source\MQTTnet\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
<!-- .NET Standard 2.0 -->
<file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\netstandard2.0\"/>
</files>
</package>
+27 -22
View File
@@ -3,35 +3,31 @@
<metadata>
<id>MQTTnet</id>
<version>0.0.0</version>
<authors>Christian Kratky</authors>
<authors>The contributors of MQTTnet</authors>
<owners>Christian Kratky</owners>
<license type="file">LICENSE</license>
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<icon>images\Logo_128x128.png</icon>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<requireLicenseAcceptance>true</requireLicenseAcceptance>
<description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
<releaseNotes>
* [All] Due to a merge issue not all changes are included in 3.0.8. All these changes are now included in this version.
* [Core] Updated all nuget references.
* [Core] Added MqttApplicationMessage.GetUserProperty() convenience method (thanks to @PMExtra).
* [LowLevelMqttClient] Added low level MQTT client in order to provide more flexibility when using the MQTT protocol. This client requires detailed knowledge about the MQTT protocol.
* [Client] Improve connection stability (thanks to @jltjohanlindqvist).
* [Client] Support WithConnectionUri to configure client (thanks to @PMExtra).
* [Client] Support PublishAsync with QoS 1 and QoS 2 from within an ApplicationMessageReceivedHandler (#648, #587, thanks to @PSanetra).
* [Client] Fixed MqttCommunicationTimedOutExceptions, caused by a long running ApplicationMessageReceivedHandler, which blocked MQTT packets from being processed (#829, thanks to @PSanetra).
* [ManagedClient] Added builder class for MqttClientUnsubscribeOptions (thanks to @dominikviererbe).
* [ManagedClient] Added support for persisted sessions (thansk to @PMExtra).
* [ManagedClient] Fixed a memory leak (thanks to @zawodskoj).
* [ManagedClient] Improved internal subscription management (#569, thanks to @cstichlberger).
* [ManagedClient] Refactored log messages (thanks to @cstichlberger).
* [Server] Added support for assigned client IDs (MQTTv5 only) (thanks to @bcrosnier).
* [Server] Added interceptor for unsubscriptions.
* [Server] Removed exceptions when user properties are set with MQTT protocol version 3.1
* [Server] Added custom session items to the client status.
* [Server] Added option to check whether the server is already started properly or not.
* [MQTTnet.AspNetCore] improved compatibility with AspNetCore 3.1
* [MQTTnet.Server] Added interceptor for unsubscriptions.
* [Core] Improved logger performance (BREAKING CHANGE!).
* [Core] Renamed some topic filter relevant classes (BREAKING CHANGE!).
* [Core] Improved task management for UWP connections (thanks to @xgstation).
* [Core] Fixed broken logger which decreases overall performance.
* [Core] Fixed issue in closed socket detection (fixes high CPU load issue).
* [Client] Added method to trigger PING/PONG manually (connection check etc.).
* [Client] Added support for certificate validation callback when using Web Sockets (requires netstandard2.1+).
* [Client] Fixed a memory leak when web socket based connections trying to reconnect with an offline server.
* [Client] Fixed a memory leak when TCP based connections trying to reconnect with an offline server.
* [Client] Fixed an issue when connecting to an invalid host (format).
* [Client] Added support for user properties in CONNECT packet.
* [Client] Removed _KeepAliveSendInterval_ and improved keep alive handling (BREAKING CHANGE!).
* [ManagedClient] Added method to trigger PING/PONG manually (connection check etc.).
* [MQTTnet.AspNetCore] improved compatibility with AspNetCore 3.1.
* [MQTTnet.AspNetCore] Fixed several packaging issues with the Nuget package.
* [MQTTnet.Server] Fixed wrong version output.
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2020</copyright>
<tags>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</tags>
@@ -48,6 +44,12 @@
<dependency id="System.Net.WebSockets" version="4.3.0" />
<dependency id="System.Net.WebSockets.Client" version="4.3.2" />
</group>
<group targetFramework="netstandard2.1">
<dependency id="NETStandard.Library" version="2.0.0" />
<dependency id="System.Net.Security" version="4.3.2" />
<dependency id="System.Net.WebSockets" version="4.3.0" />
<dependency id="System.Net.WebSockets.Client" version="4.3.2" />
</group>
<group targetFramework="uap10.0">
<dependency id="Microsoft.NETCore.UniversalWindowsPlatform" version="6.2.10" />
</group>
@@ -67,6 +69,9 @@
<!-- .NET Standard 2.0 -->
<file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\netstandard2.0\"/>
<!-- .NET Standard 2.1 -->
<file src="..\Source\MQTTnet\bin\Release\netstandard2.1\MQTTnet.*" target="lib\netstandard2.1\"/>
<!-- Universal Windows -->
<file src="..\Source\MQTTnet\bin\Release\uap10.0\MQTTnet.*" target="lib\uap10.0\"/>
+5 -1
View File
@@ -28,16 +28,19 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
# Build the ASP.NET Core 2.0 extension
&$msbuild ..\Source\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp3.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
# Build the RPC extension
&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
# Build the Managed Client extension
@@ -45,6 +48,7 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
# Build the WebSocket4Net extension
@@ -52,6 +56,7 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
&$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
# Create NuGet packages.
@@ -69,7 +74,6 @@ Copy-Item MQTTnet.Extensions.WebSocket4Net.nuspec -Destination MQTTnet.Extension
New-Item -ItemType Directory -Force -Path .\NuGet
.\nuget.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
.\nuget.exe pack MQTTnet.NETStandard.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
.\nuget.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
.\nuget.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
.\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.5 KiB

After

Width:  |  Height:  |  Size: 2.7 KiB

+1 -1
View File
@@ -1,6 +1,6 @@
MIT License
MQTTnet Copyright (c) 2016-2019 Christian Kratky
Copyright (c) 2016-2019 Christian Kratky
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
+1 -1
View File
@@ -20,7 +20,6 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1
Build\MQTTnet.Extensions.ManagedClient.nuspec = Build\MQTTnet.Extensions.ManagedClient.nuspec
Build\MQTTnet.Extensions.Rpc.nuspec = Build\MQTTnet.Extensions.Rpc.nuspec
Build\MQTTnet.Extensions.WebSocket4Net.nuspec = Build\MQTTnet.Extensions.WebSocket4Net.nuspec
Build\MQTTnet.NETStandard.nuspec = Build\MQTTnet.NETStandard.nuspec
Build\MQTTnet.nuspec = Build\MQTTnet.nuspec
Build\upload.ps1 = Build\upload.ps1
EndProjectSection
@@ -28,6 +27,7 @@ EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}"
ProjectSection(SolutionItems) = preProject
.bettercodehub.yml = .bettercodehub.yml
.editorconfig = .editorconfig
appveyor.yml = appveyor.yml
LICENSE = LICENSE
README.md = README.md
+3 -3
View File
@@ -26,7 +26,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
* Uniform API across all supported versions of the MQTT protocol
* Interfaces included for mocking and testing
* Access to internal trace messages
* Unit tested (~210 tests)
* Unit tested (~250 tests)
* No external dependencies
\* 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_.
@@ -34,7 +34,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
### Client
* Communication via TCP (+TLS) or WS (WebSocket) supported
* Included core _MqttClient_ with low level functionality
* Included core _LowLevelMqttClient_ with low level functionality
* Also included _ManagedMqttClient_ which maintains the connection and subscriptions automatically. Also application messages are queued and re-scheduled for higher QoS levels automatically.
* Rx support (via another project)
* Compatible with Microsoft Azure IoT Hub
@@ -110,7 +110,7 @@ If you use this library and want to see your project here please create a pull r
MIT License
MQTTnet Copyright (c) 2016-2019 Christian Kratky
MQTTnet Copyright (c) 2016-2020 Christian Kratky
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@@ -1,7 +1,6 @@
using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Client.Tcp;
using MQTTnet.Client.Options;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using System;
using System.Net;
@@ -10,7 +9,7 @@ namespace MQTTnet.AspNetCore.Client
{
public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
{
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
@@ -43,7 +43,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
#if NETCOREAPP3_1
public override ValueTask DisposeAsync()
#else
public new Task DisposeAsync()
public Task DisposeAsync()
#endif
{
IsConnected = false;
@@ -16,15 +16,10 @@
<ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp3.1'">
<FrameworkReference Include="Microsoft.AspNetCore.App" />
<PackageReference Include="System.IO.Pipelines" Version="4.7.1" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)' != 'netcoreapp3.1'">
<PackageReference Include="Microsoft.AspNetCore.Connections.Abstractions" Version="3.1.3" />
<PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.1.0" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.2.1" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.3" />
<PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.0.0" />
</ItemGroup>
<ItemGroup>
@@ -25,7 +25,7 @@ namespace MQTTnet.AspNetCore
_input = Connection.Transport.Input;
_output = Connection.Transport.Output;
}
_reader = new SpanBasedMqttPacketBodyReader();
}
@@ -37,6 +37,12 @@ namespace MQTTnet.AspNetCore
{
get
{
#if NETCOREAPP3_1
if (Connection?.RemoteEndPoint != null)
{
return Connection.RemoteEndPoint.ToString();
}
#endif
var connection = Http?.HttpContext?.Connection;
if (connection == null)
{
@@ -154,19 +160,19 @@ namespace MQTTnet.AspNetCore
public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken)
{
var formatter = PacketFormatterAdapter;
var buffer = formatter.Encode(packet);
var msg = buffer.AsMemory();
var output = _output;
await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
var buffer = formatter.Encode(packet);
var msg = buffer.AsMemory();
var output = _output;
var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
if (result.IsCompleted)
{
BytesSent += msg.Length;
}
PacketFormatterAdapter.FreeBuffer();
}
finally
@@ -14,7 +14,7 @@ namespace MQTTnet.AspNetCore
private readonly IMqttServerOptions _options;
public MqttHostedServer(IMqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
: base(adapters, logger.CreateChildLogger(nameof(MqttHostedServer)))
: base(adapters, logger)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
}
@@ -10,15 +10,13 @@ using System.Threading.Tasks;
namespace MQTTnet.AspNetCore
{
public class MqttWebSocketServerAdapter : IMqttServerAdapter
public sealed class MqttWebSocketServerAdapter : IMqttServerAdapter
{
private readonly IMqttNetLogger _logger;
readonly IMqttNetLogger _rootLogger;
public MqttWebSocketServerAdapter(IMqttNetLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
_rootLogger = logger ?? throw new ArgumentNullException(nameof(logger));
}
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
@@ -51,7 +49,7 @@ namespace MQTTnet.AspNetCore
var formatter = new MqttPacketFormatterAdapter(writer);
var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection, clientCertificate);
using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter))))
using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _rootLogger))
{
await clientHandler(channelAdapter).ConfigureAwait(false);
}
@@ -1,8 +1,9 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Extensions.ManagedClient
{
@@ -21,11 +22,12 @@ namespace MQTTnet.Extensions.ManagedClient
IConnectingFailedHandler ConnectingFailedHandler { get; set; }
ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; }
Task StartAsync(IManagedMqttClientOptions options);
Task StopAsync();
Task PingAsync(CancellationToken cancellationToken);
Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters);
Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters);
Task UnsubscribeAsync(IEnumerable<string> topics);
Task PublishAsync(ManagedMqttApplicationMessage applicationMessages);
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0</TargetFrameworks>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' AND '$(SolutionName)' != 'MQTTnet.noUWP' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<Product />
@@ -14,7 +14,7 @@
<CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
<NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion>10.0.17134.0</TargetPlatformVersion>
<TargetPlatformVersion>10.0.18362.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
@@ -33,7 +33,7 @@ namespace MQTTnet.Extensions.ManagedClient
private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
private readonly IMqttClient _mqttClient;
private readonly IMqttNetLogger _logger;
private readonly IMqttNetScopedLogger _logger;
private readonly AsyncLock _messageQueueLock = new AsyncLock();
@@ -48,7 +48,7 @@ namespace MQTTnet.Extensions.ManagedClient
_mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(ManagedMqttClient));
_logger = logger.CreateScopedLogger(nameof(ManagedMqttClient));
}
public bool IsConnected => _mqttClient.IsConnected;
@@ -129,6 +129,11 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
public Task PingAsync(CancellationToken cancellationToken)
{
return _mqttClient.PingAsync(cancellationToken);
}
public async Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
{
ThrowIfDisposed();
@@ -199,7 +204,7 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
public Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters)
{
ThrowIfDisposed();
@@ -449,12 +454,12 @@ namespace MQTTnet.Extensions.ManagedClient
var endTime = DateTime.UtcNow + timeout;
while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
{
List<TopicFilter> subscriptions;
List<MqttTopicFilter> subscriptions;
HashSet<string> unsubscriptions;
lock (_subscriptions)
{
subscriptions = _subscriptions.Select(i => new TopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }).ToList();
subscriptions = _subscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }).ToList();
_subscriptions.Clear();
unsubscriptions = new HashSet<string>(_unsubscriptions);
_unsubscriptions.Clear();
@@ -504,7 +509,7 @@ namespace MQTTnet.Extensions.ManagedClient
{
if (_reconnectSubscriptions.Any())
{
var subscriptions = _reconnectSubscriptions.Select(i => new TopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value });
var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value });
await _mqttClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false);
}
}
@@ -1,12 +1,12 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Extensions.ManagedClient
{
@@ -108,7 +108,7 @@ namespace MQTTnet.Extensions.ManagedClient
return client;
}
public static Task SubscribeAsync(this IManagedMqttClient client, params TopicFilter[] topicFilters)
public static Task SubscribeAsync(this IManagedMqttClient client, params MqttTopicFilter[] topicFilters)
{
if (client == null) throw new ArgumentNullException(nameof(client));
@@ -120,7 +120,7 @@ namespace MQTTnet.Extensions.ManagedClient
if (client == null) throw new ArgumentNullException(nameof(client));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
}
public static Task SubscribeAsync(this IManagedMqttClient client, string topic)
@@ -128,7 +128,7 @@ namespace MQTTnet.Extensions.ManagedClient
if (client == null) throw new ArgumentNullException(nameof(client));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).Build());
}
public static Task UnsubscribeAsync(this IManagedMqttClient client, params string[] topicFilters)
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0</TargetFrameworks>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' AND '$(SolutionName)' != 'MQTTnet.noUWP' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<Product />
@@ -14,7 +14,7 @@
<CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
<NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion>10.0.17134.0</TargetPlatformVersion>
<TargetPlatformVersion>10.0.18362.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0</TargetFrameworks>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' AND '$(SolutionName)' != 'MQTTnet.noUWP' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<Product />
@@ -14,7 +14,7 @@
<CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
<NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion>10.0.17134.0</TargetPlatformVersion>
<TargetPlatformVersion>10.0.18362.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
@@ -8,7 +8,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
{
if (mqttFactory == null) throw new ArgumentNullException(nameof(mqttFactory));
return mqttFactory.UseClientAdapterFactory(new WebSocket4NetMqttClientAdapterFactory());
return mqttFactory.UseClientAdapterFactory(new WebSocket4NetMqttClientAdapterFactory(mqttFactory.DefaultLogger));
}
}
}
@@ -9,21 +9,27 @@ namespace MQTTnet.Extensions.WebSocket4Net
{
public class WebSocket4NetMqttClientAdapterFactory : IMqttClientAdapterFactory
{
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
readonly IMqttNetLogger _logger;
public WebSocket4NetMqttClientAdapterFactory(IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (logger == null) throw new ArgumentNullException(nameof(logger));
switch (options.ChannelOptions)
{
case MqttClientTcpOptions _:
{
return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), _logger);
}
case MqttClientWebSocketOptions webSocketOptions:
{
return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), _logger);
}
default:
@@ -1,5 +1,6 @@
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using System.Diagnostics;
using System.Reflection;
namespace MQTTnet.Server.Controllers
@@ -12,7 +13,8 @@ namespace MQTTnet.Server.Controllers
[HttpGet]
public ActionResult<string> GetVersion()
{
return Assembly.GetExecutingAssembly().GetCustomAttribute<AssemblyInformationalVersionAttribute>().InformationalVersion;
var fileVersion = FileVersionInfo.GetVersionInfo(Assembly.GetExecutingAssembly().Location);
return fileVersion.ProductVersion.ToString();
}
}
}
@@ -5,7 +5,7 @@ using System.Threading;
namespace MQTTnet.Server.Logging
{
public class MqttNetLoggerWrapper : IMqttNetLogger
public sealed class MqttNetLoggerWrapper : IMqttNetLogger
{
readonly ILogger<MqttServer> _logger;
@@ -16,9 +16,9 @@ namespace MQTTnet.Server.Logging
public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
public IMqttNetLogger CreateChildLogger(string source)
public IMqttNetScopedLogger CreateScopedLogger(string source)
{
return new MqttNetLogger(source);
return new MqttNetScopedLogger(this, source);
}
public void Publish(MqttNetLogLevel level, string source, string message, object[] parameters, Exception exception)
+3 -3
View File
@@ -19,7 +19,7 @@
</PropertyGroup>
<PropertyGroup>
<StartupObject></StartupObject>
<StartupObject>MQTTnet.Server.Program</StartupObject>
<UserSecretsId>c564f0de-28b4-45bf-b726-4d665d705653</UserSecretsId>
</PropertyGroup>
@@ -47,8 +47,8 @@
<PackageReference Include="IronPython.StdLib" Version="2.7.10-candidate1" />
<PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="3.1.3" />
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="5.3.1" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="5.3.1" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="5.3.3" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="5.3.3" />
</ItemGroup>
<ItemGroup>
+5 -1
View File
@@ -2,6 +2,7 @@
using Microsoft.Extensions.Hosting;
using MQTTnet.Server.Web;
using System;
using System.Diagnostics;
using System.Reflection;
namespace MQTTnet.Server
@@ -58,8 +59,11 @@ namespace MQTTnet.Server
Console.WriteLine(@"https://github.com/chkr1011/MQTTnet");
Console.ForegroundColor = ConsoleColor.White;
var fileVersion = FileVersionInfo.GetVersionInfo(Assembly.GetExecutingAssembly().Location);
Console.WriteLine($@"
Version: {Assembly.GetExecutingAssembly().GetCustomAttribute<AssemblyInformationalVersionAttribute>().InformationalVersion}
Version: {fileVersion.ProductVersion}
License: MIT (read LICENSE file)
Sponsoring: https://opencollective.com/mqttnet
Support: https://github.com/chkr1011/MQTTnet/issues
@@ -1,10 +1,9 @@
using MQTTnet.Client.Options;
using MQTTnet.Diagnostics;
namespace MQTTnet.Adapter
{
public interface IMqttClientAdapterFactory
{
IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger);
IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options);
}
}
+7 -8
View File
@@ -16,10 +16,10 @@ namespace MQTTnet.Adapter
{
public sealed class MqttChannelAdapter : Disposable, IMqttChannelAdapter
{
const uint ErrorOperationAborted = 0x800703E3;
const int ReadBufferSize = 4096; // TODO: Move buffer size to config
const uint _errorOperationAborted = 0x800703E3;
const int _readBufferSize = 4096; // TODO: Move buffer size to config
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly IMqttChannel _channel;
readonly MqttPacketReader _packetReader;
@@ -39,7 +39,7 @@ namespace MQTTnet.Adapter
_packetReader = new MqttPacketReader(_channel);
_logger = logger.CreateChildLogger(nameof(MqttChannelAdapter));
_logger = logger.CreateScopedLogger(nameof(MqttChannelAdapter));
}
public string Endpoint => _channel.Endpoint;
@@ -128,8 +128,6 @@ namespace MQTTnet.Adapter
Interlocked.Add(ref _bytesReceived, packetData.Count);
PacketFormatterAdapter.FreeBuffer();
_logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet);
}
catch (Exception exception)
@@ -143,6 +141,7 @@ namespace MQTTnet.Adapter
}
finally
{
PacketFormatterAdapter.FreeBuffer();
_writerSemaphore?.Release();
}
}
@@ -246,7 +245,7 @@ namespace MQTTnet.Adapter
var body = new byte[fixedHeader.RemainingLength];
var bodyOffset = 0;
var chunkSize = Math.Min(ReadBufferSize, fixedHeader.RemainingLength);
var chunkSize = Math.Min(_readBufferSize, fixedHeader.RemainingLength);
do
{
@@ -305,7 +304,7 @@ namespace MQTTnet.Adapter
if (exception is COMException comException)
{
if ((uint)comException.HResult == ErrorOperationAborted)
if ((uint)comException.HResult == _errorOperationAborted)
{
throw new OperationCanceledException();
}
+7 -6
View File
@@ -1,12 +1,12 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Client.Options;
using MQTTnet.Client.Subscribing;
using MQTTnet.Client.Unsubscribing;
using System;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Client
{
@@ -16,11 +16,12 @@ namespace MQTTnet.Client
IMqttClientOptions Options { get; }
IMqttClientConnectedHandler ConnectedHandler { get; set; }
IMqttClientDisconnectedHandler DisconnectedHandler { get; set; }
Task<MqttClientAuthenticateResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken);
Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken);
Task PingAsync(CancellationToken cancellationToken);
Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken);
Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken);
+39 -33
View File
@@ -29,7 +29,7 @@ namespace MQTTnet.Client
readonly object _disconnectLock = new object();
readonly IMqttClientAdapterFactory _adapterFactory;
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
CancellationTokenSource _backgroundCancellationTokenSource;
Task _packetReceiverTask;
@@ -40,14 +40,15 @@ namespace MQTTnet.Client
IMqttChannelAdapter _adapter;
bool _cleanDisconnectInitiated;
long _disconnectGate;
long _isDisconnectPending;
bool _isConnected;
public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_adapterFactory = channelFactory ?? throw new ArgumentNullException(nameof(channelFactory));
_logger = logger.CreateChildLogger(nameof(MqttClient));
_logger = logger.CreateScopedLogger(nameof(MqttClient));
}
public IMqttClientConnectedHandler ConnectedHandler { get; set; }
@@ -56,7 +57,13 @@ namespace MQTTnet.Client
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
public bool IsConnected { get; private set; }
public bool IsConnected
{
get
{
return _isConnected && Interlocked.Read(ref _isDisconnectPending) == 0;
}
}
public IMqttClientOptions Options { get; private set; }
@@ -81,8 +88,8 @@ namespace MQTTnet.Client
_backgroundCancellationTokenSource = new CancellationTokenSource();
var backgroundCancellationToken = _backgroundCancellationTokenSource.Token;
_disconnectGate = 0;
var adapter = _adapterFactory.CreateClientAdapter(options, _logger);
_isDisconnectPending = 0;
var adapter = _adapterFactory.CreateClientAdapter(options);
_adapter = adapter;
using (var combined = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken))
@@ -107,7 +114,7 @@ namespace MQTTnet.Client
_keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(backgroundCancellationToken), backgroundCancellationToken);
}
IsConnected = true;
_isConnected = true;
_logger.Info("Connected.");
@@ -157,6 +164,11 @@ namespace MQTTnet.Client
}
}
public async Task PingAsync(CancellationToken cancellationToken)
{
await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
}
public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken)
{
if (data == null) throw new ArgumentNullException(nameof(data));
@@ -283,7 +295,10 @@ namespace MQTTnet.Client
void ThrowIfNotConnected()
{
if (!IsConnected || Interlocked.Read(ref _disconnectGate) == 1) throw new MqttCommunicationException("The client is not connected.");
if (!IsConnected || Interlocked.Read(ref _isDisconnectPending) == 1)
{
throw new MqttCommunicationException("The client is not connected.");
}
}
void ThrowIfConnected(string message)
@@ -296,7 +311,7 @@ namespace MQTTnet.Client
var clientWasConnected = IsConnected;
TryInitiateDisconnect();
IsConnected = false;
_isConnected = false;
try
{
@@ -321,7 +336,7 @@ namespace MQTTnet.Client
await Task.WhenAll(receiverTask, publishPacketReceiverTask, keepAliveTask).ConfigureAwait(false);
_publishPacketReceiverQueue.Dispose();
_publishPacketReceiverQueue?.Dispose();
}
catch (Exception e)
{
@@ -364,12 +379,9 @@ namespace MQTTnet.Client
}
}
private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken)
Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
if (cancellationToken.IsCancellationRequested)
{
return Task.FromResult(0);
}
cancellationToken.ThrowIfCancellationRequested();
_sendTracker.Restart();
@@ -379,19 +391,19 @@ namespace MQTTnet.Client
async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket
{
cancellationToken.ThrowIfCancellationRequested();
_sendTracker.Restart();
ushort identifier = 0;
if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier.HasValue)
{
identifier = packetWithIdentifier.PacketIdentifier.Value;
}
using (var packetAwaiter = _packetDispatcher.AddPacketAwaiter<TResponsePacket>(identifier))
using (var packetAwaiter = _packetDispatcher.AddAwaiter<TResponsePacket>(identifier))
{
try
{
_sendTracker.Restart();
await _adapter.SendPacketAsync(requestPacket, Options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
}
catch (Exception e)
@@ -402,9 +414,7 @@ namespace MQTTnet.Client
try
{
var response = await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false);
_receiveTracker.Restart();
return response;
return await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false);
}
catch (Exception exception)
{
@@ -424,23 +434,19 @@ namespace MQTTnet.Client
{
_logger.Verbose("Start sending keep alive packets.");
var keepAlivePeriod = Options.KeepAlivePeriod;
while (!cancellationToken.IsCancellationRequested)
{
// Values described here: [MQTT-3.1.2-24].
var keepAliveSendInterval = TimeSpan.FromSeconds(Options.KeepAlivePeriod.TotalSeconds * 0.75);
if (Options.KeepAliveSendInterval.HasValue)
{
keepAliveSendInterval = Options.KeepAliveSendInterval.Value;
}
var waitTimeSend = keepAliveSendInterval - _sendTracker.Elapsed;
var waitTimeReceive = keepAliveSendInterval - _receiveTracker.Elapsed;
if (waitTimeSend <= TimeSpan.Zero || waitTimeReceive <= TimeSpan.Zero)
var waitTime = keepAlivePeriod - _sendTracker.Elapsed;
if (waitTime <= TimeSpan.Zero)
{
await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
}
await Task.Delay(keepAliveSendInterval, cancellationToken).ConfigureAwait(false);
await Task.Delay(keepAlivePeriod, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception exception)
@@ -749,7 +755,7 @@ namespace MQTTnet.Client
bool DisconnectIsPending()
{
return Interlocked.CompareExchange(ref _disconnectGate, 1, 0) != 0;
return Interlocked.CompareExchange(ref _isDisconnectPending, 1, 0) != 0;
}
}
}
@@ -131,7 +131,7 @@ namespace MQTTnet.Client
return client.DisconnectAsync(new MqttClientDisconnectOptions());
}
public static Task<MqttClientSubscribeResult> SubscribeAsync(this IMqttClient client, params TopicFilter[] topicFilters)
public static Task<MqttClientSubscribeResult> SubscribeAsync(this IMqttClient client, params MqttTopicFilter[] topicFilters)
{
if (client == null) throw new ArgumentNullException(nameof(client));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -147,7 +147,7 @@ namespace MQTTnet.Client
if (client == null) throw new ArgumentNullException(nameof(client));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
}
public static Task<MqttClientSubscribeResult> SubscribeAsync(this IMqttClient client, string topic)
@@ -155,7 +155,7 @@ namespace MQTTnet.Client
if (client == null) throw new ArgumentNullException(nameof(client));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).Build());
}
public static Task<MqttClientUnsubscribeResult> UnsubscribeAsync(this IMqttClient client, params string[] topicFilters)
@@ -1,6 +1,8 @@
using System;
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using System;
using System.Collections.Generic;
namespace MQTTnet.Client.Options
{
@@ -12,11 +14,9 @@ namespace MQTTnet.Client.Options
IMqttExtendedAuthenticationExchangeHandler ExtendedAuthenticationExchangeHandler { get; }
MqttProtocolVersion ProtocolVersion { get; }
IMqttClientChannelOptions ChannelOptions { get; }
TimeSpan CommunicationTimeout { get; }
TimeSpan KeepAlivePeriod { get; }
TimeSpan? KeepAliveSendInterval { get; }
MqttApplicationMessage WillMessage { get; }
uint? WillDelayInterval { get; }
@@ -28,5 +28,6 @@ namespace MQTTnet.Client.Options
bool? RequestResponseInformation { get; }
uint? SessionExpiryInterval { get; }
ushort? TopicAliasMaximum { get; }
List<MqttUserProperty> UserProperties { get; set; }
}
}
@@ -0,0 +1,16 @@
using System.Net.Security;
using System.Security.Cryptography.X509Certificates;
namespace MQTTnet.Client.Options
{
public class MqttClientCertificateValidationCallbackContext
{
public X509Certificate Certificate { get; set; }
public X509Chain Chain { get; set; }
public SslPolicyErrors SslPolicyErrors { get; set; }
public IMqttClientChannelOptions ClientOptions { get; set; }
}
}
@@ -1,6 +1,8 @@
using System;
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using System;
using System.Collections.Generic;
namespace MQTTnet.Client.Options
{
@@ -15,8 +17,7 @@ namespace MQTTnet.Client.Options
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 MqttApplicationMessage WillMessage { get; set; }
public uint? WillDelayInterval { get; set; }
@@ -29,5 +30,6 @@ namespace MQTTnet.Client.Options
public bool? RequestResponseInformation { get; set; }
public uint? SessionExpiryInterval { get; set; }
public ushort? TopicAliasMaximum { get; set; }
public List<MqttUserProperty> UserProperties { get; set; }
}
}
@@ -1,6 +1,8 @@
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
@@ -38,6 +40,12 @@ namespace MQTTnet.Client.Options
return this;
}
[Obsolete("This method is no longer supported. The client will send ping requests just before the keep alive interval is going to elapse. As per MQTT RFC the serve has to wait 1.5 times the interval so we don't need this anymore.")]
public MqttClientOptionsBuilder WithKeepAliveSendInterval(TimeSpan value)
{
return this;
}
public MqttClientOptionsBuilder WithNoKeepAlive()
{
return WithKeepAlivePeriod(TimeSpan.Zero);
@@ -49,12 +57,6 @@ namespace MQTTnet.Client.Options
return this;
}
public MqttClientOptionsBuilder WithKeepAliveSendInterval(TimeSpan value)
{
_options.KeepAliveSendInterval = value;
return this;
}
public MqttClientOptionsBuilder WithClientId(string value)
{
_options.ClientId = value;
@@ -116,6 +118,20 @@ namespace MQTTnet.Client.Options
return this;
}
public MqttClientOptionsBuilder WithUserProperty(string name, string value)
{
if (name is null) throw new ArgumentNullException(nameof(name));
if (value is null) throw new ArgumentNullException(nameof(value));
if (_options.UserProperties == null)
{
_options.UserProperties = new List<MqttUserProperty>();
}
_options.UserProperties.Add(new MqttUserProperty(name, value));
return this;
}
public MqttClientOptionsBuilder WithCredentials(string username, string password = null)
{
byte[] passwordBuffer = null;
@@ -261,7 +277,11 @@ namespace MQTTnet.Client.Options
#else
Certificates = _tlsParameters.Certificates?.ToList(),
#endif
#pragma warning disable CS0618 // Type or member is obsolete
CertificateValidationCallback = _tlsParameters.CertificateValidationCallback,
#pragma warning restore CS0618 // Type or member is obsolete
CertificateValidationHandler = _tlsParameters.CertificateValidationHandler,
IgnoreCertificateChainErrors = _tlsParameters.IgnoreCertificateChainErrors,
IgnoreCertificateRevocationErrors = _tlsParameters.IgnoreCertificateRevocationErrors
};
@@ -10,12 +10,15 @@ namespace MQTTnet.Client.Options
{
public bool UseTls { get; set; }
[Obsolete("This property will be removed soon. Use CertificateValidationHandler instead.")]
public Func<X509Certificate, X509Chain, SslPolicyErrors, IMqttClientOptions, bool> CertificateValidationCallback
{
get;
set;
}
public Func<MqttClientCertificateValidationCallbackContext, bool> CertificateValidationHandler { get; set; }
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12;
#if WINDOWS_UWP
@@ -23,7 +26,6 @@ namespace MQTTnet.Client.Options
#else
public IEnumerable<X509Certificate> Certificates { get; set; }
#endif
public bool AllowUntrustedCertificates { get; set; }
@@ -15,6 +15,7 @@ namespace MQTTnet.Client.Options
public bool IgnoreCertificateChainErrors { get; set; }
public bool AllowUntrustedCertificates { get; set; }
#if WINDOWS_UWP
public List<byte[]> Certificates { get; set; }
#else
@@ -23,6 +24,9 @@ namespace MQTTnet.Client.Options
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12;
[Obsolete("This property will be removed soon. Use CertificateValidationHandler instead.")]
public Func<X509Certificate, X509Chain, SslPolicyErrors, IMqttClientOptions, bool> CertificateValidationCallback { get; set; }
public Func<MqttClientCertificateValidationCallbackContext, bool> CertificateValidationHandler { get; set; }
}
}
@@ -5,7 +5,7 @@ namespace MQTTnet.Client.Subscribing
{
public class MqttClientSubscribeOptions
{
public List<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
public List<MqttTopicFilter> TopicFilters { get; set; } = new List<MqttTopicFilter>();
public uint? SubscriptionIdentifier { get; set; }
@@ -1,7 +1,7 @@
using System;
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
namespace MQTTnet.Client.Subscribing
{
@@ -38,7 +38,7 @@ namespace MQTTnet.Client.Subscribing
bool? retainAsPublished = null,
MqttRetainHandling? retainHandling = null)
{
return WithTopicFilter(new TopicFilter
return WithTopicFilter(new MqttTopicFilter
{
Topic = topic,
QualityOfServiceLevel = qualityOfServiceLevel,
@@ -48,30 +48,30 @@ namespace MQTTnet.Client.Subscribing
});
}
public MqttClientSubscribeOptionsBuilder WithTopicFilter(Action<TopicFilterBuilder> topicFilterBuilder)
public MqttClientSubscribeOptionsBuilder WithTopicFilter(Action<MqttTopicFilterBuilder> topicFilterBuilder)
{
if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder));
var internalTopicFilterBuilder = new TopicFilterBuilder();
var internalTopicFilterBuilder = new MqttTopicFilterBuilder();
topicFilterBuilder(internalTopicFilterBuilder);
return WithTopicFilter(internalTopicFilterBuilder);
}
public MqttClientSubscribeOptionsBuilder WithTopicFilter(TopicFilterBuilder topicFilterBuilder)
public MqttClientSubscribeOptionsBuilder WithTopicFilter(MqttTopicFilterBuilder topicFilterBuilder)
{
if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder));
return WithTopicFilter(topicFilterBuilder.Build());
}
public MqttClientSubscribeOptionsBuilder WithTopicFilter(TopicFilter topicFilter)
public MqttClientSubscribeOptionsBuilder WithTopicFilter(MqttTopicFilter topicFilter)
{
if (topicFilter == null) throw new ArgumentNullException(nameof(topicFilter));
if (_subscribeOptions.TopicFilters == null)
{
_subscribeOptions.TopicFilters = new List<TopicFilter>();
_subscribeOptions.TopicFilters = new List<MqttTopicFilter>();
}
_subscribeOptions.TopicFilters.Add(topicFilter);
@@ -4,13 +4,13 @@ namespace MQTTnet.Client.Subscribing
{
public class MqttClientSubscribeResultItem
{
public MqttClientSubscribeResultItem(TopicFilter topicFilter, MqttClientSubscribeResultCode resultCode)
public MqttClientSubscribeResultItem(MqttTopicFilter topicFilter, MqttClientSubscribeResultCode resultCode)
{
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
ResultCode = resultCode;
}
public TopicFilter TopicFilter { get; }
public MqttTopicFilter TopicFilter { get; }
public MqttClientSubscribeResultCode ResultCode { get; }
}
@@ -44,7 +44,7 @@ namespace MQTTnet.Client.Unsubscribing
return this;
}
public MqttClientUnsubscribeOptionsBuilder WithTopicFilter(TopicFilter topicFilter)
public MqttClientUnsubscribeOptionsBuilder WithTopicFilter(MqttTopicFilter topicFilter)
{
if (topicFilter is null) throw new ArgumentNullException(nameof(topicFilter));
+2 -2
View File
@@ -6,8 +6,8 @@ namespace MQTTnet.Diagnostics
{
event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
IMqttNetLogger CreateChildLogger(string source);
IMqttNetScopedLogger CreateScopedLogger(string source);
void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception);
void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception);
}
}
@@ -0,0 +1,11 @@
using System;
namespace MQTTnet.Diagnostics
{
public interface IMqttNetScopedLogger
{
IMqttNetScopedLogger CreateScopedLogger(string source);
void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception);
}
}
+12 -41
View File
@@ -5,16 +5,7 @@ namespace MQTTnet.Diagnostics
public class MqttNetLogger : IMqttNetLogger
{
readonly string _logId;
readonly string _source;
readonly MqttNetLogger _parentLogger;
public MqttNetLogger(string source, string logId)
{
_source = source;
_logId = logId;
}
public MqttNetLogger()
{
}
@@ -24,52 +15,41 @@ namespace MQTTnet.Diagnostics
_logId = logId;
}
MqttNetLogger(MqttNetLogger parentLogger, string logId, string source)
{
_parentLogger = parentLogger ?? throw new ArgumentNullException(nameof(parentLogger));
_source = source ?? throw new ArgumentNullException(nameof(source));
_logId = logId;
}
public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
// TODO: Consider creating a LoggerFactory which will allow creating loggers. The logger factory will
// be the only place which has the published event.
public IMqttNetLogger CreateChildLogger(string source)
public IMqttNetScopedLogger CreateScopedLogger(string source)
{
if (source is null) throw new ArgumentNullException(nameof(source));
return new MqttNetLogger(this, _logId, source);
return new MqttNetScopedLogger(this, source);
}
public void Publish(MqttNetLogLevel level, string message, object[] parameters, Exception exception)
public void Publish(MqttNetLogLevel level, string source, string message, object[] parameters, Exception exception)
{
var hasLocalListeners = LogMessagePublished != null;
var hasGlobalListeners = MqttNetGlobalLogger.HasListeners;
if (!hasLocalListeners && !hasGlobalListeners && _parentLogger == null)
if (!hasLocalListeners && !hasGlobalListeners)
{
return;
}
if (parameters?.Length > 0)
try
{
try
{
message = string.Format(message, parameters);
}
catch
{
message = "MESSAGE FORMAT INVALID: " + message;
}
message = string.Format(message ?? string.Empty, parameters);
}
catch (FormatException)
{
message = "MESSAGE FORMAT INVALID: " + message;
}
var logMessage = new MqttNetLogMessage
{
LogId = _logId,
Timestamp = DateTime.UtcNow,
Source = _source,
Source = source,
ThreadId = Environment.CurrentManagedThreadId,
Level = level,
Message = message,
@@ -85,15 +65,6 @@ namespace MQTTnet.Diagnostics
{
LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage));
}
_parentLogger?.Publish(logMessage);
}
void Publish(MqttNetLogMessage logMessage)
{
LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage));
_parentLogger?.Publish(logMessage);
}
}
}
@@ -1,35 +0,0 @@
using System;
namespace MQTTnet.Diagnostics
{
public static class MqttNetLoggerExtensions
{
public static void Verbose(this IMqttNetLogger logger, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));
logger.Publish(MqttNetLogLevel.Verbose, message, parameters, null);
}
public static void Info(this IMqttNetLogger logger, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));
logger.Publish(MqttNetLogLevel.Info, message, parameters, null);
}
public static void Warning(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));
logger.Publish(MqttNetLogLevel.Warning, message, parameters, exception);
}
public static void Error(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));
logger.Publish(MqttNetLogLevel.Error, message, parameters, exception);
}
}
}
@@ -0,0 +1,26 @@
using System;
namespace MQTTnet.Diagnostics
{
public sealed class MqttNetScopedLogger : IMqttNetScopedLogger
{
readonly IMqttNetLogger _logger;
readonly string _source;
public MqttNetScopedLogger(IMqttNetLogger logger, string source)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_source = source ?? throw new ArgumentNullException(nameof(source));
}
public IMqttNetScopedLogger CreateScopedLogger(string source)
{
return new MqttNetScopedLogger(_logger, source);
}
public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception)
{
_logger.Publish(logLevel, _source, message, parameters, exception);
}
}
}
@@ -0,0 +1,27 @@
using System;
namespace MQTTnet.Diagnostics
{
public static class MqttNetScopedLoggerExtensions
{
public static void Verbose(this IMqttNetScopedLogger logger, string message, params object[] parameters)
{
logger.Publish(MqttNetLogLevel.Verbose, message, parameters, null);
}
public static void Info(this IMqttNetScopedLogger logger, string message, params object[] parameters)
{
logger.Publish(MqttNetLogLevel.Info, message, parameters, null);
}
public static void Warning(this IMqttNetScopedLogger logger, Exception exception, string message, params object[] parameters)
{
logger.Publish(MqttNetLogLevel.Warning, message, parameters, exception);
}
public static void Error(this IMqttNetScopedLogger logger, Exception exception, string message, params object[] parameters)
{
logger.Publish(MqttNetLogLevel.Error, message, parameters, exception);
}
}
}
@@ -16,6 +16,8 @@
return "netstandard1.3";
#elif NETSTANDARD2_0
return "netstandard2.0";
#elif NETSTANDARD2_1
return "netstandard2.1";
#elif WINDOWS_UWP
return "uap10.0";
#endif
@@ -1,4 +1,5 @@
using System;
using System.Runtime.CompilerServices;
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Formatter.V3;
@@ -9,12 +10,7 @@ namespace MQTTnet.Formatter
{
public class MqttPacketFormatterAdapter
{
private IMqttPacketFormatter _formatter;
public MqttPacketFormatterAdapter()
: this(new MqttPacketWriter())
{
}
IMqttPacketFormatter _formatter;
public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion)
: this(protocolVersion, new MqttPacketWriter())
@@ -120,7 +116,7 @@ namespace MQTTnet.Formatter
}
}
private MqttProtocolVersion ParseProtocolVersion(ReceivedMqttPacket receivedMqttPacket)
MqttProtocolVersion ParseProtocolVersion(ReceivedMqttPacket receivedMqttPacket)
{
if (receivedMqttPacket == null) throw new ArgumentNullException(nameof(receivedMqttPacket));
@@ -155,7 +151,8 @@ namespace MQTTnet.Formatter
throw new MqttProtocolViolationException($"Protocol '{protocolName}' not supported.");
}
private void ThrowIfFormatterNotSet()
[MethodImpl(MethodImplOptions.AggressiveInlining)]
void ThrowIfFormatterNotSet()
{
if (_formatter == null)
{
+9 -9
View File
@@ -11,17 +11,17 @@ namespace MQTTnet.Formatter
/// same as for the original MemoryStream in .net. Also this implementation allows accessing the internal
/// buffer for all platforms and .net framework versions (which is not available at the regular MemoryStream).
/// </summary>
public class MqttPacketWriter : IMqttPacketWriter
public sealed class MqttPacketWriter : IMqttPacketWriter
{
private static readonly ArraySegment<byte> ZeroVariableLengthIntegerArray = new ArraySegment<byte>(new byte[1], 0, 1);
private static readonly ArraySegment<byte> ZeroTwoByteIntegerArray = new ArraySegment<byte>(new byte[2], 0, 2);
static readonly ArraySegment<byte> ZeroVariableLengthIntegerArray = new ArraySegment<byte>(new byte[1], 0, 1);
static readonly ArraySegment<byte> ZeroTwoByteIntegerArray = new ArraySegment<byte>(new byte[2], 0, 2);
public static int InitialBufferSize = 128;
public static int MaxBufferSize = 4096;
private byte[] _buffer = new byte[InitialBufferSize];
byte[] _buffer = new byte[InitialBufferSize];
private int _offset;
int _offset;
public int Length { get; private set; }
@@ -191,13 +191,13 @@ namespace MQTTnet.Formatter
Array.Resize(ref _buffer, MaxBufferSize);
}
private void Write(ArraySegment<byte> buffer)
void Write(ArraySegment<byte> buffer)
{
Write(buffer.Array, buffer.Offset, buffer.Count);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void EnsureAdditionalCapacity(int additionalCapacity)
void EnsureAdditionalCapacity(int additionalCapacity)
{
var freeSpace = _buffer.Length - _offset;
if (freeSpace >= additionalCapacity)
@@ -209,7 +209,7 @@ namespace MQTTnet.Formatter
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void EnsureCapacity(int capacity)
void EnsureCapacity(int capacity)
{
var newBufferLength = _buffer.Length;
@@ -227,7 +227,7 @@ namespace MQTTnet.Formatter
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void IncreasePosition(int length)
void IncreasePosition(int length)
{
_offset += length;
@@ -1,5 +1,6 @@
using System;
using System.Linq;
using System.Runtime.CompilerServices;
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
@@ -9,26 +10,19 @@ namespace MQTTnet.Formatter.V3
{
public class MqttV310PacketFormatter : IMqttPacketFormatter
{
private const int FixedHeaderSize = 1;
const int FixedHeaderSize = 1;
private static readonly MqttPingReqPacket PingReqPacket = new MqttPingReqPacket();
private static readonly MqttPingRespPacket PingRespPacket = new MqttPingRespPacket();
private static readonly MqttDisconnectPacket DisconnectPacket = new MqttDisconnectPacket();
static readonly MqttPingReqPacket PingReqPacket = new MqttPingReqPacket();
static readonly MqttPingRespPacket PingRespPacket = new MqttPingRespPacket();
static readonly MqttDisconnectPacket DisconnectPacket = new MqttDisconnectPacket();
private readonly IMqttPacketWriter _packetWriter;
public MqttV310PacketFormatter()
: this(new MqttPacketWriter())
{
}
readonly IMqttPacketWriter _packetWriter;
public MqttV310PacketFormatter(IMqttPacketWriter packetWriter)
{
_packetWriter = packetWriter;
}
public IMqttDataConverter DataConverter { get; } = new MqttV310DataConverter();
public ArraySegment<byte> Encode(MqttBasePacket packet)
@@ -93,7 +87,7 @@ namespace MQTTnet.Formatter.V3
_packetWriter.FreeBuffer();
}
private byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter)
byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter)
{
switch (packet)
{
@@ -116,7 +110,7 @@ namespace MQTTnet.Formatter.V3
}
}
private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -126,7 +120,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -136,7 +130,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -146,7 +140,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -156,7 +150,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -166,7 +160,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -183,7 +177,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -194,7 +188,7 @@ namespace MQTTnet.Formatter.V3
while (!body.EndOfStream)
{
var topicFilter = new TopicFilter
var topicFilter = new MqttTopicFilter
{
Topic = body.ReadStringWithLengthPrefix(),
QualityOfServiceLevel = (MqttQualityOfServiceLevel)body.ReadByte()
@@ -206,7 +200,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
private static MqttBasePacket DecodePublishPacket(ReceivedMqttPacket receivedMqttPacket)
static MqttBasePacket DecodePublishPacket(ReceivedMqttPacket receivedMqttPacket)
{
ThrowIfBodyIsEmpty(receivedMqttPacket.Body);
@@ -239,7 +233,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
private MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body)
MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -300,7 +294,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body)
static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -340,7 +334,7 @@ namespace MQTTnet.Formatter.V3
}
// ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
private static void ValidatePublishPacket(MqttPublishPacket packet)
static void ValidatePublishPacket(MqttPublishPacket packet)
{
if (packet.QualityOfServiceLevel == 0 && packet.Dup)
{
@@ -418,7 +412,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
private static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter)
static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -430,7 +424,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
}
private static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter)
static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter)
{
ValidatePublishPacket(packet);
@@ -475,7 +469,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
}
private static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter)
static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -487,7 +481,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
}
private static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter)
static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -499,7 +493,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
}
private static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter)
static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -511,7 +505,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
}
private static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter)
static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
@@ -534,7 +528,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02);
}
private static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter)
static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -554,7 +548,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck);
}
private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter)
static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
@@ -576,7 +570,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
}
private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter)
static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -587,12 +581,12 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck);
}
private static byte EncodeEmptyPacket(MqttControlPacketType type)
static byte EncodeEmptyPacket(MqttControlPacketType type)
{
return MqttPacketWriter.BuildFixedHeader(type);
}
// ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
[MethodImpl(MethodImplOptions.AggressiveInlining)]
protected static void ThrowIfBodyIsEmpty(IMqttPacketBodyReader body)
{
if (body == null || body.Length == 0)
@@ -6,11 +6,6 @@ namespace MQTTnet.Formatter.V3
{
public class MqttV311PacketFormatter : MqttV310PacketFormatter
{
public MqttV311PacketFormatter()
: base()
{
}
public MqttV311PacketFormatter(IMqttPacketWriter packetWriter)
: base(packetWriter)
{
@@ -1,7 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Publishing;
@@ -11,7 +8,9 @@ using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using MqttClientSubscribeResult = MQTTnet.Client.Subscribing.MqttClientSubscribeResult;
namespace MQTTnet.Formatter.V5
@@ -127,7 +126,8 @@ namespace MQTTnet.Formatter.V5
RequestProblemInformation = options.RequestProblemInformation,
RequestResponseInformation = options.RequestResponseInformation,
SessionExpiryInterval = options.SessionExpiryInterval,
TopicAliasMaximum = options.TopicAliasMaximum
TopicAliasMaximum = options.TopicAliasMaximum,
UserProperties = options.UserProperties
}
};
}
@@ -244,11 +244,11 @@ namespace MQTTnet.Formatter.V5
{
// QoS 0 has no response. So we treat it as a success always.
// Both enums have the same values. So it can be easily converted.
result.ReasonCode = (MqttClientPublishReasonCode)pubAckPacket.ReasonCode;
result.ReasonCode = (MqttClientPublishReasonCode)pubAckPacket.ReasonCode;
result.PacketIdentifier = pubAckPacket.PacketIdentifier;
result.PacketIdentifier = pubAckPacket.PacketIdentifier;
}
return result;
}
@@ -405,7 +405,7 @@ namespace MQTTnet.Formatter.V5
var retainAsPublished = (options & (1 << 3)) > 0;
var retainHandling = (MqttRetainHandling)((options >> 4) & 3);
packet.TopicFilters.Add(new TopicFilter
packet.TopicFilters.Add(new MqttTopicFilter
{
Topic = topic,
QualityOfServiceLevel = qos,
@@ -1,4 +1,5 @@
using System;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;
@@ -10,6 +11,8 @@ namespace MQTTnet.Implementations
{
readonly Socket _socket;
NetworkStream _networkStream;
public CrossPlatformSocket(AddressFamily addressFamily)
{
_socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
@@ -25,6 +28,7 @@ namespace MQTTnet.Implementations
public CrossPlatformSocket(Socket socket)
{
_socket = socket ?? throw new ArgumentNullException(nameof(socket));
_networkStream = new NetworkStream(socket, true);
}
public bool NoDelay
@@ -137,7 +141,9 @@ namespace MQTTnet.Implementations
try
{
// Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430
_networkStream?.Dispose();
// Workaround for: https://github.com/dotnet/corefx/issues/24430
using (cancellationToken.Register(() => _socket.Dispose()))
{
cancellationToken.ThrowIfCancellationRequested();
@@ -147,6 +153,7 @@ namespace MQTTnet.Implementations
#else
await _socket.ConnectAsync(host, port).ConfigureAwait(false);
#endif
_networkStream = new NetworkStream(_socket, true);
}
}
catch (ObjectDisposedException)
@@ -190,11 +197,18 @@ namespace MQTTnet.Implementations
public NetworkStream GetStream()
{
return new NetworkStream(_socket, true);
var networkStream = _networkStream;
if (networkStream == null)
{
throw new IOException("The socket is not connected.");
}
return networkStream;
}
public void Dispose()
{
_networkStream?.Dispose();
_socket?.Dispose();
}
@@ -8,7 +8,14 @@ namespace MQTTnet.Implementations
{
public class MqttClientAdapterFactory : IMqttClientAdapterFactory
{
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
readonly IMqttNetLogger _logger;
public MqttClientAdapterFactory(IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
@@ -16,12 +23,12 @@ namespace MQTTnet.Implementations
{
case MqttClientTcpOptions _:
{
return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttPacketWriter()), _logger);
}
case MqttClientWebSocketOptions webSocketOptions:
{
return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttPacketWriter()), _logger);
}
default:
@@ -64,7 +64,7 @@ namespace MQTTnet.Implementations
if (_options.TlsOptions?.UseTls != true)
{
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString());
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString()).AsTask().ConfigureAwait(false);
}
else
{
@@ -85,7 +85,7 @@ namespace MQTTnet.Implementations
socketProtectionLevel = SocketProtectionLevel.Tls10;
}
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), socketProtectionLevel);
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), socketProtectionLevel).AsTask().ConfigureAwait(false);
}
Endpoint = _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;
@@ -46,54 +46,68 @@ namespace MQTTnet.Implementations
public async Task ConnectAsync(CancellationToken cancellationToken)
{
CrossPlatformSocket socket;
if (_options.AddressFamily == AddressFamily.Unspecified)
CrossPlatformSocket socket = null;
try
{
socket = new CrossPlatformSocket();
}
else
{
socket = new CrossPlatformSocket(_options.AddressFamily);
}
socket.ReceiveBufferSize = _options.BufferSize;
socket.SendBufferSize = _options.BufferSize;
socket.NoDelay = _options.NoDelay;
if (_options.DualMode.HasValue)
{
// It is important to avoid setting the flag if no specific value is set by the user
// because on IPv4 only networks the setter will always throw an exception. Regardless
// of the actual value.
socket.DualMode = _options.DualMode.Value;
}
await socket.ConnectAsync(_options.Server, _options.GetPort(), cancellationToken).ConfigureAwait(false);
var networkStream = socket.GetStream();
if (_options.TlsOptions.UseTls)
{
var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
try
if (_options.AddressFamily == AddressFamily.Unspecified)
{
await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
socket = new CrossPlatformSocket();
}
catch
else
{
sslStream.Dispose();
throw;
socket = new CrossPlatformSocket(_options.AddressFamily);
}
_stream = sslStream;
}
else
{
_stream = networkStream;
}
socket.ReceiveBufferSize = _options.BufferSize;
socket.SendBufferSize = _options.BufferSize;
socket.NoDelay = _options.NoDelay;
Endpoint = socket.RemoteEndPoint?.ToString();
if (_options.DualMode.HasValue)
{
// It is important to avoid setting the flag if no specific value is set by the user
// because on IPv4 only networks the setter will always throw an exception. Regardless
// of the actual value.
socket.DualMode = _options.DualMode.Value;
}
await socket.ConnectAsync(_options.Server, _options.GetPort(), cancellationToken).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
var networkStream = socket.GetStream();
if (_options.TlsOptions?.UseTls == true)
{
var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
try
{
await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
}
catch
{
#if NETSTANDARD2_1
await sslStream.DisposeAsync().ConfigureAwait(false);
#else
sslStream.Dispose();
#endif
throw;
}
_stream = sslStream;
}
else
{
_stream = networkStream;
}
Endpoint = socket.RemoteEndPoint?.ToString();
}
catch (Exception)
{
socket?.Dispose();
throw;
}
}
public Task DisconnectAsync(CancellationToken cancellationToken)
@@ -106,19 +120,20 @@ namespace MQTTnet.Implementations
{
if (buffer is null) throw new ArgumentNullException(nameof(buffer));
cancellationToken.ThrowIfCancellationRequested();
try
{
// Workaround for: https://github.com/dotnet/corefx/issues/24430
using (cancellationToken.Register(Dispose))
{
cancellationToken.ThrowIfCancellationRequested();
return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
}
}
catch (ObjectDisposedException)
{
return -1;
// Indicate a graceful socket close.
return 0;
}
catch (IOException exception)
{
@@ -135,13 +150,13 @@ namespace MQTTnet.Implementations
{
if (buffer is null) throw new ArgumentNullException(nameof(buffer));
cancellationToken.ThrowIfCancellationRequested();
try
{
// Workaround for: https://github.com/dotnet/corefx/issues/24430
using (cancellationToken.Register(Dispose))
{
cancellationToken.ThrowIfCancellationRequested();
await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
}
}
@@ -181,9 +196,30 @@ namespace MQTTnet.Implementations
bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
{
if (_options.TlsOptions.CertificateValidationCallback != null)
#region OBSOLETE
#pragma warning disable CS0618 // Type or member is obsolete
var certificateValidationCallback = _options?.TlsOptions?.CertificateValidationCallback;
#pragma warning restore CS0618 // Type or member is obsolete
if (certificateValidationCallback != null)
{
return _options.TlsOptions.CertificateValidationCallback(x509Certificate, chain, sslPolicyErrors, _clientOptions);
return certificateValidationCallback(x509Certificate, chain, sslPolicyErrors, _clientOptions);
}
#endregion
var certificateValidationHandler = _options?.TlsOptions?.CertificateValidationHandler;
if (certificateValidationHandler != null)
{
var context = new MqttClientCertificateValidationCallbackContext
{
Certificate = x509Certificate,
Chain = chain,
SslPolicyErrors = sslPolicyErrors,
ClientOptions = _options
};
return certificateValidationHandler(context);
}
if (sslPolicyErrors == SslPolicyErrors.None)
@@ -13,16 +13,16 @@ namespace MQTTnet.Implementations
{
public sealed class MqttTcpServerAdapter : IMqttServerAdapter
{
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly IMqttNetLogger _rootLogger;
IMqttServerOptions _options;
StreamSocketListener _listener;
public MqttTcpServerAdapter(IMqttNetLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
_rootLogger = logger ?? throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateScopedLogger(nameof(MqttTcpServerAdapter));
}
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
@@ -89,7 +89,7 @@ namespace MQTTnet.Implementations
}
}
using (var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, clientCertificate, _options), new MqttPacketFormatterAdapter(), _logger))
using (var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, clientCertificate, _options), new MqttPacketFormatterAdapter(new MqttPacketWriter()), _rootLogger))
{
await clientHandler(clientAdapter).ConfigureAwait(false);
}
@@ -16,15 +16,15 @@ namespace MQTTnet.Implementations
public sealed class MqttTcpServerAdapter : Disposable, IMqttServerAdapter
{
readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly IMqttNetLogger _rootLogger;
CancellationTokenSource _cancellationTokenSource;
public MqttTcpServerAdapter(IMqttNetLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
_rootLogger = logger ?? throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateScopedLogger(nameof(MqttTcpServerAdapter));
}
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
@@ -105,11 +105,7 @@ namespace MQTTnet.Implementations
{
if (!options.BoundInterNetworkAddress.Equals(IPAddress.None))
{
var listenerV4 = new MqttTcpServerListener(
AddressFamily.InterNetwork,
options,
tlsCertificate,
_logger)
var listenerV4 = new MqttTcpServerListener(AddressFamily.InterNetwork, options, tlsCertificate, _rootLogger)
{
ClientHandler = OnClientAcceptedAsync
};
@@ -122,11 +118,7 @@ namespace MQTTnet.Implementations
if (!options.BoundInterNetworkV6Address.Equals(IPAddress.None))
{
var listenerV6 = new MqttTcpServerListener(
AddressFamily.InterNetworkV6,
options,
tlsCertificate,
_logger)
var listenerV6 = new MqttTcpServerListener(AddressFamily.InterNetworkV6, options, tlsCertificate, _rootLogger)
{
ClientHandler = OnClientAcceptedAsync
};
@@ -17,7 +17,8 @@ namespace MQTTnet.Implementations
{
public sealed class MqttTcpServerListener : IDisposable
{
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly IMqttNetLogger _rootLogger;
readonly AddressFamily _addressFamily;
readonly MqttServerTcpEndpointBaseOptions _options;
readonly MqttServerTlsTcpEndpointOptions _tlsOptions;
@@ -35,7 +36,8 @@ namespace MQTTnet.Implementations
_addressFamily = addressFamily;
_options = options;
_tlsCertificate = tlsCertificate;
_logger = logger.CreateChildLogger(nameof(MqttTcpServerListener));
_rootLogger = logger;
_logger = logger.CreateScopedLogger(nameof(MqttTcpServerListener));
if (_options is MqttServerTlsTcpEndpointOptions tlsOptions)
{
@@ -178,7 +180,7 @@ namespace MQTTnet.Implementations
var clientHandler = ClientHandler;
if (clientHandler != null)
{
using (var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(stream, remoteEndPoint, clientCertificate), new MqttPacketFormatterAdapter(), _logger))
using (var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(stream, remoteEndPoint, clientCertificate), new MqttPacketFormatterAdapter(new MqttPacketWriter()), _rootLogger))
{
await clientHandler(clientAdapter).ConfigureAwait(false);
}
@@ -1,21 +1,21 @@
using System;
using MQTTnet.Channel;
using MQTTnet.Client.Options;
using MQTTnet.Internal;
using System;
using System.Net;
using System.Net.WebSockets;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Channel;
using MQTTnet.Client.Options;
using MQTTnet.Internal;
namespace MQTTnet.Implementations
{
public class MqttWebSocketChannel : Disposable, IMqttChannel
public sealed class MqttWebSocketChannel : IMqttChannel
{
private readonly MqttClientWebSocketOptions _options;
readonly MqttClientWebSocketOptions _options;
private SemaphoreSlim _sendLock = new SemaphoreSlim(1, 1);
private WebSocket _webSocket;
AsyncLock _sendLock = new AsyncLock();
WebSocket _webSocket;
public MqttWebSocketChannel(MqttClientWebSocketOptions options)
{
@@ -53,6 +53,67 @@ namespace MQTTnet.Implementations
}
var clientWebSocket = new ClientWebSocket();
try
{
SetupClientWebSocket(clientWebSocket);
await clientWebSocket.ConnectAsync(new Uri(uri), cancellationToken).ConfigureAwait(false);
}
catch (Exception)
{
// Prevent a memory leak when always creating new instance which will fail while connecting.
clientWebSocket.Dispose();
throw;
}
_webSocket = clientWebSocket;
IsSecureConnection = uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase);
}
public async Task DisconnectAsync(CancellationToken cancellationToken)
{
if (_webSocket == null)
{
return;
}
if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting)
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
}
Cleanup();
}
public async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, offset, count), cancellationToken).ConfigureAwait(false);
return response.Count;
}
public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
// The lock is required because the client will throw an exception if _SendAsync_ is
// called from multiple threads at the same time. But this issue only happens with several
// framework versions.
if (_sendLock == null)
{
return;
}
using (await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false))
{
await _webSocket.SendAsync(new ArraySegment<byte>(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken).ConfigureAwait(false);
}
}
public void Dispose()
{
Cleanup();
}
void SetupClientWebSocket(ClientWebSocket clientWebSocket)
{
if (_options.ProxyOptions != null)
{
@@ -90,68 +151,37 @@ namespace MQTTnet.Implementations
#else
clientWebSocket.Options.ClientCertificates.Add(certificate);
#endif
}
}
await clientWebSocket.ConnectAsync(new Uri(uri), cancellationToken).ConfigureAwait(false);
_webSocket = clientWebSocket;
var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler;
#if NETSTANDARD2_1
if (certificateValidationHandler != null)
{
clientWebSocket.Options.RemoteCertificateValidationCallback = new System.Net.Security.RemoteCertificateValidationCallback((sender, certificate, chain, sslPolicyErrors) =>
{
// TODO: Find a way to add client options to same callback. Problem is that they have a different type.
var context = new MqttClientCertificateValidationCallbackContext
{
Certificate = certificate,
Chain = chain,
SslPolicyErrors = sslPolicyErrors,
ClientOptions = _options
};
IsSecureConnection = uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase);
return certificateValidationHandler(context);
});
}
#else
if (certificateValidationHandler != null)
{
throw new NotSupportedException("The remote certificate validation callback for Web Sockets is only supported for netstandard 2.1+");
}
#endif
}
public async Task DisconnectAsync(CancellationToken cancellationToken)
{
if (_webSocket == null)
{
return;
}
if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting)
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
}
Cleanup();
}
public async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, offset, count), cancellationToken).ConfigureAwait(false);
return response.Count;
}
public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
// The lock is required because the client will throw an exception if _SendAsync_ is
// called from multiple threads at the same time. But this issue only happens with several
// framework versions.
if (_sendLock == null)
{
return;
}
await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await _webSocket.SendAsync(new ArraySegment<byte>(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken).ConfigureAwait(false);
}
finally
{
_sendLock?.Release();
}
}
protected override void Dispose(bool disposing)
{
if (disposing)
{
Cleanup();
}
base.Dispose(disposing);
}
private void Cleanup()
void Cleanup()
{
_sendLock?.Dispose();
_sendLock = null;
@@ -169,7 +199,7 @@ namespace MQTTnet.Implementations
}
}
private IWebProxy CreateProxy()
IWebProxy CreateProxy()
{
if (string.IsNullOrEmpty(_options.ProxyOptions?.Address))
{
@@ -1,23 +0,0 @@
using System;
namespace MQTTnet.Internal
{
public static class ArraySegmentExtensions
{
public static byte[] ToArray(this ArraySegment<byte> source)
{
if (source.Array == null)
{
return null;
}
var buffer = new byte[source.Count];
if (buffer.Length > 0)
{
Array.Copy(source.Array, source.Offset, buffer, 0, buffer.Length);
}
return buffer;
}
}
}
+22 -20
View File
@@ -4,11 +4,12 @@ using System.Threading;
namespace MQTTnet.Internal
{
public class BlockingQueue<TItem> : Disposable
public sealed class BlockingQueue<TItem> : IDisposable
{
private readonly object _syncRoot = new object();
private readonly LinkedList<TItem> _items = new LinkedList<TItem>();
private readonly ManualResetEventSlim _gate = new ManualResetEventSlim(false);
readonly object _syncRoot = new object();
readonly LinkedList<TItem> _items = new LinkedList<TItem>();
ManualResetEventSlim _gate = new ManualResetEventSlim(false);
public int Count
{
@@ -28,13 +29,13 @@ namespace MQTTnet.Internal
lock (_syncRoot)
{
_items.AddLast(item);
_gate.Set();
_gate?.Set();
}
}
public TItem Dequeue(CancellationToken cancellationToken = default(CancellationToken))
public TItem Dequeue(CancellationToken cancellationToken = default)
{
while (true)
while (!cancellationToken.IsCancellationRequested)
{
lock (_syncRoot)
{
@@ -48,17 +49,19 @@ namespace MQTTnet.Internal
if (_items.Count == 0)
{
_gate.Reset();
_gate?.Reset();
}
}
_gate.Wait(cancellationToken);
_gate?.Wait(cancellationToken);
}
throw new OperationCanceledException();
}
public TItem PeekAndWait(CancellationToken cancellationToken = default(CancellationToken))
public TItem PeekAndWait(CancellationToken cancellationToken = default)
{
while (true)
while (!cancellationToken.IsCancellationRequested)
{
lock (_syncRoot)
{
@@ -69,12 +72,14 @@ namespace MQTTnet.Internal
if (_items.Count == 0)
{
_gate.Reset();
_gate?.Reset();
}
}
_gate.Wait(cancellationToken);
_gate?.Wait(cancellationToken);
}
throw new OperationCanceledException();
}
public void RemoveFirst(Predicate<TItem> match)
@@ -109,13 +114,10 @@ namespace MQTTnet.Internal
}
}
protected override void Dispose(bool disposing)
public void Dispose()
{
if (disposing)
{
_gate.Dispose();
}
base.Dispose(disposing);
_gate?.Dispose();
_gate = null;
}
}
}
+1 -1
View File
@@ -5,7 +5,7 @@ namespace MQTTnet.Internal
{
public static class TaskExtensions
{
public static void Forget(this Task task, IMqttNetLogger logger)
public static void Forget(this Task task, IMqttNetScopedLogger logger)
{
task?.ContinueWith(t =>
{
@@ -10,7 +10,7 @@ namespace MQTTnet.LowLevelClient
{
public sealed class LowLevelMqttClient : ILowLevelMqttClient
{
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly IMqttClientAdapterFactory _clientAdapterFactory;
IMqttChannelAdapter _adapter;
@@ -22,7 +22,7 @@ namespace MQTTnet.LowLevelClient
if (logger is null) throw new ArgumentNullException(nameof(logger));
_clientAdapterFactory = clientAdapterFactory;
_logger = logger.CreateChildLogger(nameof(LowLevelMqttClient));
_logger = logger.CreateScopedLogger(nameof(LowLevelMqttClient));
}
bool IsConnected => _adapter != null;
@@ -36,7 +36,7 @@ namespace MQTTnet.LowLevelClient
throw new InvalidOperationException("Low level MQTT client is already connected. Disconnect first before connecting again.");
}
var newAdapter = _clientAdapterFactory.CreateClientAdapter(options, _logger);
var newAdapter = _clientAdapterFactory.CreateClientAdapter(options);
try
{
+4 -8
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0</TargetFrameworks>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' AND '$(SolutionName)' != 'MQTTnet.noUWP' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<AssemblyName>MQTTnet</AssemblyName>
@@ -20,8 +20,8 @@
<CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
<NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion>10.0.17134.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10586.212</TargetPlatformMinVersion>
<TargetPlatformVersion>10.0.18362.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
@@ -54,7 +54,7 @@
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.8" />
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net452'">
@@ -62,9 +62,5 @@
<ItemGroup Condition="'$(TargetFramework)'=='net461'">
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
</ItemGroup>
</Project>
+2 -1
View File
@@ -11,7 +11,7 @@ namespace MQTTnet
{
public sealed class MqttFactory : IMqttFactory
{
IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory();
IMqttClientAdapterFactory _clientAdapterFactory;
public MqttFactory() : this(new MqttNetLogger())
{
@@ -20,6 +20,7 @@ namespace MQTTnet
public MqttFactory(IMqttNetLogger logger)
{
DefaultLogger = logger ?? throw new ArgumentNullException(nameof(logger));
_clientAdapterFactory = new MqttClientAdapterFactory(logger);
}
public IMqttNetLogger DefaultLogger { get; }
@@ -1,9 +1,14 @@
using MQTTnet.Protocol;
using System;
namespace MQTTnet
{
// TODO: Consider renaming to "MqttTopicFilter"
public class TopicFilter
[Obsolete("Use MqttTopicFilter instead. It is just a renamed version to align with general namings in this lib.")]
public class TopicFilter : MqttTopicFilter
{
}
public class MqttTopicFilter
{
public string Topic { get; set; }
+57
View File
@@ -0,0 +1,57 @@
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
using System;
namespace MQTTnet
{
[Obsolete("Use MqttTopicFilter instead. It is just a renamed version to align with general namings in this lib.")]
public class TopicFilterBuilder : MqttTopicFilterBuilder
{
}
public class MqttTopicFilterBuilder
{
MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
string _topic;
public MqttTopicFilterBuilder WithTopic(string topic)
{
_topic = topic;
return this;
}
public MqttTopicFilterBuilder WithQualityOfServiceLevel(MqttQualityOfServiceLevel qualityOfServiceLevel)
{
_qualityOfServiceLevel = qualityOfServiceLevel;
return this;
}
public MqttTopicFilterBuilder WithAtLeastOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
return this;
}
public MqttTopicFilterBuilder WithAtMostOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
return this;
}
public MqttTopicFilterBuilder WithExactlyOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce;
return this;
}
public MqttTopicFilter Build()
{
if (string.IsNullOrEmpty(_topic))
{
throw new MqttProtocolViolationException("Topic is not set.");
}
return new MqttTopicFilter { Topic = _topic, QualityOfServiceLevel = _qualityOfServiceLevel };
}
}
}
@@ -1,5 +1,5 @@
using System;
using MQTTnet.Packets;
using MQTTnet.Packets;
using System;
namespace MQTTnet.PacketDispatcher
{
@@ -1,5 +1,4 @@
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using System;
using System.Threading;
@@ -7,11 +6,11 @@ using System.Threading.Tasks;
namespace MQTTnet.PacketDispatcher
{
public sealed class MqttPacketAwaiter<TPacket> : Disposable, IMqttPacketAwaiter where TPacket : MqttBasePacket
public sealed class MqttPacketAwaiter<TPacket> : IMqttPacketAwaiter where TPacket : MqttBasePacket
{
private readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource;
private readonly ushort? _packetIdentifier;
private readonly MqttPacketDispatcher _owningPacketDispatcher;
readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource;
readonly ushort? _packetIdentifier;
readonly MqttPacketDispatcher _owningPacketDispatcher;
public MqttPacketAwaiter(ushort? packetIdentifier, MqttPacketDispatcher owningPacketDispatcher)
{
@@ -39,7 +38,6 @@ namespace MQTTnet.PacketDispatcher
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
#if NET452
// To prevent deadlocks it is required to call the _TrySetResult_ method
// from a new thread because the awaiting code will not(!) be executed in
@@ -81,14 +79,9 @@ namespace MQTTnet.PacketDispatcher
#endif
}
protected override void Dispose(bool disposing)
public void Dispose()
{
if (disposing)
{
_owningPacketDispatcher.RemovePacketAwaiter<TPacket>(_packetIdentifier);
}
base.Dispose(disposing);
_owningPacketDispatcher.RemoveAwaiter<TPacket>(_packetIdentifier);
}
}
}
@@ -1,22 +1,22 @@
using System;
using System.Collections.Concurrent;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using System;
using System.Collections.Concurrent;
namespace MQTTnet.PacketDispatcher
{
public class MqttPacketDispatcher
public sealed class MqttPacketDispatcher
{
private readonly ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter> _packetAwaiters = new ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter>();
readonly ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter> _awaiters = new ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter>();
public void Dispatch(Exception exception)
{
foreach (var awaiter in _packetAwaiters)
foreach (var awaiter in _awaiters)
{
awaiter.Value.Fail(exception);
}
_packetAwaiters.Clear();
_awaiters.Clear();
}
public void Dispatch(MqttBasePacket packet)
@@ -25,7 +25,7 @@ namespace MQTTnet.PacketDispatcher
if (packet is MqttDisconnectPacket disconnectPacket)
{
foreach (var packetAwaiter in _packetAwaiters)
foreach (var packetAwaiter in _awaiters)
{
packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
}
@@ -42,7 +42,7 @@ namespace MQTTnet.PacketDispatcher
var type = packet.GetType();
var key = new Tuple<ushort, Type>(identifier, type);
if (_packetAwaiters.TryRemove(key, out var awaiter))
if (_awaiters.TryRemove(key, out var awaiter))
{
awaiter.Complete(packet);
return;
@@ -53,15 +53,15 @@ namespace MQTTnet.PacketDispatcher
public void Reset()
{
foreach (var awaiter in _packetAwaiters)
foreach (var awaiter in _awaiters)
{
awaiter.Value.Cancel();
awaiter.Value.Cancel();
}
_packetAwaiters.Clear();
_awaiters.Clear();
}
public MqttPacketAwaiter<TResponsePacket> AddPacketAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
public MqttPacketAwaiter<TResponsePacket> AddAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
{
if (!identifier.HasValue)
{
@@ -71,7 +71,7 @@ namespace MQTTnet.PacketDispatcher
var awaiter = new MqttPacketAwaiter<TResponsePacket>(identifier, this);
var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
if (!_packetAwaiters.TryAdd(key, awaiter))
if (!_awaiters.TryAdd(key, awaiter))
{
throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{key.Item2.Name}' with identifier {key.Item1}.");
}
@@ -79,7 +79,7 @@ namespace MQTTnet.PacketDispatcher
return awaiter;
}
public void RemovePacketAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
public void RemoveAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
{
if (!identifier.HasValue)
{
@@ -87,7 +87,7 @@ namespace MQTTnet.PacketDispatcher
}
var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
_packetAwaiters.TryRemove(key, out _);
_awaiters.TryRemove(key, out _);
}
}
}
@@ -7,7 +7,7 @@ namespace MQTTnet.Packets
{
public ushort? PacketIdentifier { get; set; }
public List<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
public List<MqttTopicFilter> TopicFilters { get; set; } = new List<MqttTopicFilter>();
#region Added in MQTTv5
@@ -11,14 +11,17 @@ namespace MQTTnet.Protocol
throw new MqttProtocolViolationException("Topic should not be empty.");
}
if (topic.Contains("+"))
foreach(var @char in topic)
{
throw new MqttProtocolViolationException("The character '+' is not allowed in topics.");
}
if (@char == '+')
{
throw new MqttProtocolViolationException("The character '+' is not allowed in topics.");
}
if (topic.Contains("#"))
{
throw new MqttProtocolViolationException("The character '#' is not allowed in topics.");
if (@char == '#')
{
throw new MqttProtocolViolationException("The character '#' is not allowed in topics.");
}
}
}
}
@@ -16,6 +16,6 @@ namespace MQTTnet.Server
Task<IList<MqttApplicationMessage>> GetMessagesAsync();
Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters);
Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters);
}
}
+1 -1
View File
@@ -23,7 +23,7 @@ namespace MQTTnet.Server
Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync();
Task ClearRetainedApplicationMessagesAsync();
Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters);
Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters);
Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters);
Task StartAsync(IMqttServerOptions options);
@@ -9,7 +9,7 @@ namespace MQTTnet.Server
IDictionary<object, object> Items { get; }
IList<TopicFilter> Subscriptions { get; }
IList<MqttTopicFilter> Subscriptions { get; }
MqttApplicationMessage WillMessage { get; }
@@ -28,7 +28,7 @@ namespace MQTTnet.Server
readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
readonly MqttClientSessionsManager _sessionsManager;
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly IMqttServerOptions _serverOptions;
readonly IMqttChannelAdapter _channelAdapter;
@@ -71,9 +71,9 @@ namespace MQTTnet.Server
ConnectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientConnection));
_logger = logger.CreateScopedLogger(nameof(MqttClientConnection));
_keepAliveMonitor = new MqttClientKeepAliveMonitor(ConnectPacket.ClientId, () => StopAsync(), _logger);
_keepAliveMonitor = new MqttClientKeepAliveMonitor(ConnectPacket.ClientId, () => StopAsync(), logger);
_connectedTimestamp = DateTime.UtcNow;
_lastPacketReceivedTimestamp = _connectedTimestamp;
@@ -282,7 +282,7 @@ namespace MQTTnet.Server
_cancellationToken.Cancel(false);
}
async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
{
var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
foreach (var applicationMessage in retainedMessages)
@@ -430,14 +430,14 @@ namespace MQTTnet.Server
}
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
var awaiter = _packetDispatcher.AddPacketAwaiter<MqttPubAckPacket>(publishPacket.PacketIdentifier);
var awaiter = _packetDispatcher.AddAwaiter<MqttPubAckPacket>(publishPacket.PacketIdentifier);
await SendAsync(publishPacket).ConfigureAwait(false);
await awaiter.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
}
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{
using (var awaiter1 = _packetDispatcher.AddPacketAwaiter<MqttPubRecPacket>(publishPacket.PacketIdentifier))
using (var awaiter2 = _packetDispatcher.AddPacketAwaiter<MqttPubCompPacket>(publishPacket.PacketIdentifier))
using (var awaiter1 = _packetDispatcher.AddAwaiter<MqttPubRecPacket>(publishPacket.PacketIdentifier))
using (var awaiter2 = _packetDispatcher.AddAwaiter<MqttPubCompPacket>(publishPacket.PacketIdentifier))
{
await SendAsync(publishPacket).ConfigureAwait(false);
await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
@@ -7,13 +7,13 @@ using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttClientKeepAliveMonitor
public sealed class MqttClientKeepAliveMonitor
{
readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
readonly string _clientId;
readonly Func<Task> _keepAliveElapsedCallback;
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
bool _isPaused;
@@ -23,7 +23,7 @@ namespace MQTTnet.Server
_keepAliveElapsedCallback = keepAliveElapsedCallback ?? throw new ArgumentNullException(nameof(keepAliveElapsedCallback));
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor));
_logger = logger.CreateScopedLogger(nameof(MqttClientKeepAliveMonitor));
}
public void Start(int keepAlivePeriod, CancellationToken cancellationToken)
+10 -4
View File
@@ -8,12 +8,18 @@ namespace MQTTnet.Server
{
public class MqttClientSession
{
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly DateTime _createdTimestamp = DateTime.UtcNow;
readonly IMqttRetainedMessagesManager _retainedMessagesManager;
public MqttClientSession(string clientId, IDictionary<object, object> items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
public MqttClientSession(
string clientId,
IDictionary<object, object> items,
MqttServerEventDispatcher eventDispatcher,
IMqttServerOptions serverOptions,
IMqttRetainedMessagesManager retainedMessagesManager,
IMqttNetLogger logger)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
Items = items ?? throw new ArgumentNullException(nameof(items));
@@ -22,7 +28,7 @@ namespace MQTTnet.Server
ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientSession));
_logger = logger.CreateScopedLogger(nameof(MqttClientSession));
}
public string ClientId { get; }
@@ -53,7 +59,7 @@ namespace MQTTnet.Server
ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
}
public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters)
public async Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters)
{
if (topicFilters is null) throw new ArgumentNullException(nameof(topicFilters));
@@ -28,7 +28,8 @@ namespace MQTTnet.Server
readonly IMqttRetainedMessagesManager _retainedMessagesManager;
readonly IMqttServerOptions _options;
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
readonly IMqttNetLogger _rootLogger;
public MqttClientSessionsManager(
IMqttServerOptions options,
@@ -40,7 +41,8 @@ namespace MQTTnet.Server
_cancellationToken = cancellationToken;
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientSessionsManager));
_logger = logger.CreateScopedLogger(nameof(MqttClientSessionsManager));
_rootLogger = logger;
_eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
_options = options ?? throw new ArgumentNullException(nameof(options));
@@ -108,7 +110,7 @@ namespace MQTTnet.Server
_messageQueue.Enqueue(new MqttEnqueuedApplicationMessage(applicationMessage, sender));
}
public Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters)
public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
{
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -371,11 +373,11 @@ namespace MQTTnet.Server
if (session == null)
{
session = new MqttClientSession(connectPacket.ClientId, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _logger);
session = new MqttClientSession(connectPacket.ClientId, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
_logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
}
var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _logger);
var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _rootLogger);
_connections[connection.ClientId] = connection;
_sessions[session.ClientId] = session;
@@ -9,7 +9,7 @@ namespace MQTTnet.Server
{
public class MqttClientSubscriptionsManager
{
private readonly Dictionary<string, TopicFilter> _subscriptions = new Dictionary<string, TopicFilter>();
private readonly Dictionary<string, MqttTopicFilter> _subscriptions = new Dictionary<string, MqttTopicFilter>();
private readonly MqttClientSession _clientSession;
private readonly IMqttServerOptions _serverOptions;
private readonly MqttServerEventDispatcher _eventDispatcher;
@@ -74,7 +74,7 @@ namespace MQTTnet.Server
return result;
}
public async Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
public async Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -206,7 +206,7 @@ namespace MQTTnet.Server
}
}
private async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(TopicFilter topicFilter)
private async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(MqttTopicFilter topicFilter)
{
var context = new MqttSubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items);
if (_serverOptions.SubscriptionInterceptor != null)
@@ -10,17 +10,19 @@ namespace MQTTnet.Server
{
public class MqttRetainedMessagesManager : IMqttRetainedMessagesManager
{
private readonly byte[] _emptyArray = new byte[0];
private readonly AsyncLock _messagesLock = new AsyncLock();
private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
readonly byte[] _emptyArray = new byte[0];
readonly AsyncLock _messagesLock = new AsyncLock();
readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
private IMqttNetLogger _logger;
private IMqttServerOptions _options;
IMqttNetScopedLogger _logger;
IMqttServerOptions _options;
// TODO: Get rid of the logger here!
public Task Start(IMqttServerOptions options, IMqttNetLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttRetainedMessagesManager));
_logger = logger.CreateScopedLogger(nameof(MqttRetainedMessagesManager));
_options = options ?? throw new ArgumentNullException(nameof(options));
return PlatformAbstractionLayer.CompletedTask;
}
@@ -105,7 +107,7 @@ namespace MQTTnet.Server
}
}
public async Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters)
public async Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+8 -6
View File
@@ -17,7 +17,8 @@ namespace MQTTnet.Server
{
readonly MqttServerEventDispatcher _eventDispatcher;
readonly ICollection<IMqttServerAdapter> _adapters;
readonly IMqttNetLogger _logger;
readonly IMqttNetLogger _rootLogger;
readonly IMqttNetScopedLogger _logger;
MqttClientSessionsManager _clientSessionsManager;
IMqttRetainedMessagesManager _retainedMessagesManager;
@@ -29,9 +30,10 @@ namespace MQTTnet.Server
_adapters = adapters.ToList();
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttServer));
_logger = logger.CreateScopedLogger(nameof(MqttServer));
_rootLogger = logger;
_eventDispatcher = new MqttServerEventDispatcher(logger.CreateChildLogger(nameof(MqttServerEventDispatcher)));
_eventDispatcher = new MqttServerEventDispatcher(logger);
}
public bool IsStarted => _cancellationTokenSource != null;
@@ -87,7 +89,7 @@ namespace MQTTnet.Server
return _retainedMessagesManager.GetMessagesAsync();
}
public Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters)
public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
{
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -127,10 +129,10 @@ namespace MQTTnet.Server
_cancellationTokenSource = new CancellationTokenSource();
_retainedMessagesManager = Options.RetainedMessagesManager;
await _retainedMessagesManager.Start(Options, _logger);
await _retainedMessagesManager.Start(Options, _rootLogger);
await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
_clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _cancellationTokenSource.Token, _eventDispatcher, _logger);
_clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _cancellationTokenSource.Token, _eventDispatcher, _rootLogger);
_clientSessionsManager.Start();
foreach (var adapter in _adapters)
@@ -4,7 +4,7 @@ namespace MQTTnet.Server
{
public class MqttServerClientSubscribedTopicEventArgs : EventArgs
{
public MqttServerClientSubscribedTopicEventArgs(string clientId, TopicFilter topicFilter)
public MqttServerClientSubscribedTopicEventArgs(string clientId, MqttTopicFilter topicFilter)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
@@ -12,6 +12,6 @@ namespace MQTTnet.Server
public string ClientId { get; }
public TopicFilter TopicFilter { get; }
public MqttTopicFilter TopicFilter { get; }
}
}
@@ -5,13 +5,15 @@ using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerEventDispatcher
public sealed class MqttServerEventDispatcher
{
readonly IMqttNetLogger _logger;
readonly IMqttNetScopedLogger _logger;
public MqttServerEventDispatcher(IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
if (logger is null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateScopedLogger(nameof(MqttServerEventDispatcher));
}
public IMqttServerClientConnectedHandler ClientConnectedHandler { get; set; }
@@ -60,7 +62,7 @@ namespace MQTTnet.Server
}
}
public async Task SafeNotifyClientSubscribedTopicAsync(string clientId, TopicFilter topicFilter)
public async Task SafeNotifyClientSubscribedTopicAsync(string clientId, MqttTopicFilter topicFilter)
{
try
{
@@ -1,10 +1,10 @@
using System;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Protocol;
namespace MQTTnet.Server
{
@@ -106,7 +106,7 @@ namespace MQTTnet.Server
return server;
}
public static Task SubscribeAsync(this IMqttServer server, string clientId, params TopicFilter[] topicFilters)
public static Task SubscribeAsync(this IMqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
@@ -121,7 +121,7 @@ namespace MQTTnet.Server
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return server.SubscribeAsync(clientId, new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
}
public static Task SubscribeAsync(this IMqttServer server, string clientId, string topic)
@@ -130,7 +130,7 @@ namespace MQTTnet.Server
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return server.SubscribeAsync(clientId, new TopicFilterBuilder().WithTopic(topic).Build());
return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build());
}
public static Task UnsubscribeAsync(this IMqttServer server, string clientId, params string[] topicFilters)
@@ -4,7 +4,7 @@ namespace MQTTnet.Server
{
public class MqttSubscriptionInterceptorContext
{
public MqttSubscriptionInterceptorContext(string clientId, TopicFilter topicFilter, IDictionary<object, object> sessionItems)
public MqttSubscriptionInterceptorContext(string clientId, MqttTopicFilter topicFilter, IDictionary<object, object> sessionItems)
{
ClientId = clientId;
TopicFilter = topicFilter;
@@ -13,7 +13,7 @@ namespace MQTTnet.Server
public string ClientId { get; }
public TopicFilter TopicFilter { get; set; }
public MqttTopicFilter TopicFilter { get; set; }
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
-51
View File
@@ -1,51 +0,0 @@
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
namespace MQTTnet
{
public class TopicFilterBuilder
{
private MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
private string _topic;
public TopicFilterBuilder WithTopic(string topic)
{
_topic = topic;
return this;
}
public TopicFilterBuilder WithQualityOfServiceLevel(MqttQualityOfServiceLevel qualityOfServiceLevel)
{
_qualityOfServiceLevel = qualityOfServiceLevel;
return this;
}
public TopicFilterBuilder WithAtLeastOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
return this;
}
public TopicFilterBuilder WithAtMostOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
return this;
}
public TopicFilterBuilder WithExactlyOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce;
return this;
}
public TopicFilter Build()
{
if (string.IsNullOrEmpty(_topic))
{
throw new MqttProtocolViolationException("Topic is not set.");
}
return new TopicFilter { Topic = _topic, QualityOfServiceLevel = _qualityOfServiceLevel };
}
}
}
@@ -1,16 +1,24 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netcoreapp3.1</TargetFramework>
<TargetFrameworks>netcoreapp3.1;net461</TargetFrameworks>
<IsPackable>false</IsPackable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.5.0" />
<PackageReference Include="Microsoft.AspNetCore.Hosting" Version="2.2.7" />
<PackageReference Include="Microsoft.AspNetCore.Server.Kestrel" Version="2.2.0" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.6.0" />
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
<PackageReference Include="MSTest.TestFramework" Version="2.1.1" />
</ItemGroup>
<ItemGroup Condition=" '$(TargetFramework)' != 'netcoreapp3.1' ">
<PackageReference Include="Microsoft.AspNetCore.Hosting" Version="2.1.1" />
<PackageReference Include="Microsoft.AspNetCore.Server.Kestrel" Version="2.1.1" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
<ProjectReference Include="..\MQTTnet.Core.Tests\MQTTnet.Tests.csproj" />
@@ -0,0 +1,50 @@
using Microsoft.AspNetCore.Connections;
using MQTTnet.Adapter;
using MQTTnet.Formatter;
using MQTTnet.Server;
using System;
using System.Threading.Tasks;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class ConnectionHandlerMockup: IMqttServerAdapter
{
public TaskCompletionSource<MqttConnectionContext> Context { get; } = new TaskCompletionSource<MqttConnectionContext>();
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public ConnectionHandlerMockup()
{
}
public async Task OnConnectedAsync(ConnectionContext connection)
{
try
{
var writer = new SpanBasedMqttPacketWriter();
var formatter = new MqttPacketFormatterAdapter(writer);
var context = new MqttConnectionContext(formatter, connection);
Context.TrySetResult(context);
await ClientHandler(context);
}
catch (Exception ex)
{
Context.TrySetException(ex);
}
}
public Task StartAsync(IMqttServerOptions options)
{
return Task.CompletedTask;
}
public Task StopAsync()
{
return Task.CompletedTask;
}
public void Dispose()
{
}
}
}
@@ -1,13 +1,20 @@
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.AspNetCore.Tests.Mockups;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Tests.Mockups;
using MQTTnet.Client.Options;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using System.Net;
namespace MQTTnet.AspNetCore.Tests
{
@@ -63,5 +70,43 @@ namespace MQTTnet.AspNetCore.Tests
var readResult = await pipe.Send.Reader.ReadAsync();
Assert.IsTrue(readResult.Buffer.Length > 20000);
}
private class Startup
{
public void Configure(IApplicationBuilder app)
{
}
}
[TestMethod]
public async Task TestEndpoint()
{
var mockup = new ConnectionHandlerMockup();
using (var host = new WebHostBuilder()
.UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
.UseStartup<Startup>()
.ConfigureServices((hostContext, services) =>
{
services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint());
services.AddSingleton<IMqttServerAdapter>(mockup);
})
.Build())
using (var client = new MqttFactory().CreateMqttClient())
{
host.Start();
await client.ConnectAsync(new MqttClientOptionsBuilder()
.WithTcpServer("localhost")
.Build(), CancellationToken.None);
var ctx = await mockup.Context.Task;
#if NETCOREAPP3_1
var ep = IPEndPoint.Parse(ctx.Endpoint);
Assert.IsNotNull(ep);
#endif
Assert.IsNotNull(ctx);
}
}
}
}
@@ -43,7 +43,7 @@ namespace MQTTnet.Benchmarks
var channel = new TestMqttChannel(_stream);
_channelAdapter = new MqttChannelAdapter(channel, serializer, new MqttNetLogger().CreateChildLogger(nameof(MqttChannelAdapter)));
_channelAdapter = new MqttChannelAdapter(channel, serializer, new MqttNetLogger());
}
[Benchmark]
@@ -13,8 +13,8 @@ namespace MQTTnet.Benchmarks.Configurations
{
public AllowNonOptimized()
{
Add(JitOptimizationsValidator.DontFailOnError); // ALLOW NON-OPTIMIZED DLLS
Add(Job.InProcess);
AddValidator(JitOptimizationsValidator.DontFailOnError); // ALLOW NON-OPTIMIZED DLLS
AddJob(Job.InProcess);
}
}
}
@@ -7,9 +7,9 @@ namespace MQTTnet.Benchmarks.Configurations
{
public BaseConfig()
{
Add(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default
Add(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default
Add(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default
AddLogger(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default
AddExporter(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default
AddColumnProvider(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default
}
}
}

Some files were not shown because too many files have changed in this diff Show More