Compare commits

...
62 Commits
Author SHA1 Message Date
Christian 54a0265ef9 Fix UnitTests 2018-04-01 20:53:32 +02:00
Christian 8f96acaec4 Fixed a race condition in client and server. 2018-03-31 23:10:14 +02:00
Christian cea984cc2f Fix stream and socket handling. 2018-03-29 22:12:23 +02:00
Christian fcf748e980 Merge branch 'develop' 2018-03-25 14:45:35 +02:00
Christian e58cb3c9cd Update docs. 2018-03-25 14:42:43 +02:00
Christian 3587a8ebd1 Add new option which allows configuring the send interval for keep alive packets. 2018-03-25 14:42:34 +02:00
Christian 3193381b69 Refactor web socket stream. Add mqtt as default sub protocol. 2018-03-25 14:42:12 +02:00
Christian 7a9ddfc2b5 Fix packet identifier which is sent from the server. 2018-03-25 14:41:12 +02:00
Christian 447b2aa2c6 Add new extension methods. 2018-03-25 14:39:44 +02:00
Christian b3eaeea8a4 Fix log issue in UWP test app. 2018-03-25 14:38:36 +02:00
Christian 45b4c05867 Update version 2018-03-25 14:38:16 +02:00
Christian 49da66dcff Fix issue when disconnecting on UWP 2018-03-13 23:28:50 +01:00
Christian bb9d5586f0 Update nuspec 2018-03-13 20:31:01 +01:00
Christian c5d22f4b27 Update docs 2018-03-13 20:23:36 +01:00
Christian e57ec11ed8 Update libs 2018-03-13 20:23:31 +01:00
Christian 56acc82445 Fix null ref in server when using interceptor 2018-03-13 20:23:22 +01:00
Christian 6e828d091f Override ToString() for LogMessage 2018-03-13 20:23:07 +01:00
Christian e1d8697331 Fix wrong parsing and generating of MqttConnAck for version 3.1.0 2018-03-13 20:22:48 +01:00
Christian 90098c2006 Optimize dispatcher 2018-03-13 20:22:13 +01:00
Christian b3a25fb9e8 Fix subscription handling of _ManagedClient_. 2018-03-05 20:43:37 +01:00
Christian f85ad4965b Refactor keep alive monitoring and add unit tests. 2018-02-28 21:43:19 +01:00
Christian f5696b6fc0 Merge branch 'develop' 2018-02-21 21:33:01 +01:00
Christian c26555aca5 Refactor ManagedClient 2018-02-21 21:32:21 +01:00
Christian 0accb467b5 Merge branch 'develop' of https://github.com/chkr1011/MQTTnet into develop 2018-02-21 20:44:10 +01:00
Christian 3b34fd4750 Fix build 2018-02-21 20:44:05 +01:00
Christian abb216c055 Fix retained message handling 2018-02-21 20:43:22 +01:00
Christian 1760dc7d35 Fix async issues 2018-02-21 20:43:09 +01:00
Christian 034e7fd750 Add event for processed application message (managed client only) 2018-02-21 20:42:39 +01:00
Christian 76c7b319b8 Fix build errors 2018-02-21 20:42:05 +01:00
Christian 09010e95a2 Merge pull request #180 from lerppana/managed_unsubscribe_fix
Fix ManagedMqttClient missing call to unsubscribe topics
2018-02-21 20:09:10 +01:00
Lassi Laati 2341d66192 Fix ManagedMqttClient missing call to unsubscribe topics 2018-02-20 09:23:47 +02:00
Christian 21586b7710 Merge branch 'master' of https://github.com/chkr1011/MQTTnet 2018-02-04 20:10:11 +01:00
Christian b426ae1198 Update docs 2018-02-04 20:06:01 +01:00
Christian e9db9d65ac Update README.md 2018-02-04 20:05:34 +01:00
Christian 59dcbf4c0c Add server events for subscribed and unsubscribed topics. Add pending messages count for connected clients. 2018-02-02 22:44:21 +01:00
Christian a204d233f6 Update configs 2018-02-02 00:47:00 +01:00
Christian 26cd84c88c Fix generation of packet identifier. 2018-02-02 00:46:47 +01:00
Christian 5f77255893 Update docs 2018-01-27 14:12:54 +01:00
Christian 8856a7aa34 Fix broken package reader 2018-01-26 23:56:22 +01:00
Christian 30754433eb Add server methods for controlling the client subscriptions. 2018-01-26 22:06:21 +01:00
Christian 751ac19c4b Fix reading more data than required for the body. 2018-01-26 21:25:56 +01:00
Christian 1415e4b878 Covnert the body buffer to a memory stream which will grow automatically. 2018-01-24 20:49:47 +01:00
Christian e56203faac Update docs 2018-01-24 20:16:47 +01:00
Christian 64c997a4a2 Merge pull request #163 from kpreisser/fixInfiniteLoop
Fix infinite loop when the client closes the connection
2018-01-24 20:12:15 +01:00
Christian 391ee34a0f Merge pull request #166 from kpreisser/fixDecodeRemainingLength
Fix the algorithm for decoding the body length
2018-01-24 20:09:14 +01:00
kpreisser 88de5047bc Fix the algorithm for decoding the remaining length to allow a length greater than 2,097,151 bytes.
See: https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os.html#_Toc442180732
2018-01-24 15:32:36 +01:00
kpreisser d020e809a1 Fix infinite loop that can occur when the client closes the connection before sending the complete packet body. 2018-01-23 10:45:58 +01:00
Christian 8b2b8d41df Update docs 2018-01-22 21:11:44 +01:00
Christian 40c33ac639 Refactoring 2018-01-22 21:11:35 +01:00
Christian 002e4b8d69 Fix broken unit tests 2018-01-22 20:55:50 +01:00
Christian c060c379d3 Add thrown exception for disconnected event 2018-01-22 20:55:06 +01:00
Christian 4d13407a83 Merge branch 'develop' of https://github.com/chkr1011/MQTTnet into develop 2018-01-21 14:33:33 +01:00
Christian d737f550c7 Add encrypted code signing key 2018-01-21 14:33:29 +01:00
Christian 5e8d5b9219 Add documents 2018-01-21 14:29:43 +01:00
Christian f98f21ffdf Update libs 2018-01-21 14:29:36 +01:00
Christian ec9e64ebfb Update docs 2018-01-21 14:14:04 +01:00
Christian 39672970e2 Update 3rd-Partly libs 2018-01-21 14:13:50 +01:00
Christian 5becf7980a Merge pull request #157 from kpreisser/trueAsync
Use asynchronous methods for blocking operations
2018-01-21 14:12:29 +01:00
Christian c7c84ad39b Make clients interfaces disposable. 2018-01-21 14:10:23 +01:00
Christian 8773117493 Fix an issue which adds a delay after every accepted connection. 2018-01-21 14:09:34 +01:00
kpreisser bb697e1bd9 Follow-Up: Add missing ConfigureAwait() call. 2018-01-17 11:29:48 +01:00
kpreisser e1a9aa7d03 Adjust the MqttClientPendingMessagesQueue and the MqttPacketReader to use asynchronous methods for long-running/blocking operations.
Otherwise, Worker Threads from the .NET Threadpool are blocked while waiting for the next client message, and while waiting for the next message to send.
2018-01-17 11:10:39 +01:00
76 changed files with 1646 additions and 635 deletions
+4 -4
View File
@@ -2,7 +2,7 @@
<package >
<metadata>
<id>MQTTnet.AspNetCore</id>
<version>2.6.0</version>
<version>2.7.3</version>
<authors>Christian Kratky</authors>
<owners>Christian Kratky</owners>
<licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,13 +10,13 @@
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<description>This is a support library to integrate MQTTnet into AspNetCore.</description>
<releaseNotes>* Updated to MQTTnet 2.6.0.
<releaseNotes>* Updated to MQTTnet 2.7.3.
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2017</copyright>
<copyright>Copyright Christian Kratky 2016-2018</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="netstandard2.0">
<dependency id="MQTTnet" version="2.6.0" />
<dependency id="MQTTnet" version="2.7.3" />
</group>
</dependencies>
</metadata>
+10 -23
View File
@@ -2,7 +2,7 @@
<package >
<metadata>
<id>MQTTnet</id>
<version>2.6.0</version>
<version>2.7.3</version>
<authors>Christian Kratky</authors>
<owners>Christian Kratky</owners>
<licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,28 +10,16 @@
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
<releaseNotes>* [Core] Merged projects (BREAKING CHANGE! But only namespace changes).
* [Core] Added a strong name for the assembly.
* [Core] Performance optimizations.
* [Core] Fixed a logging issue when dealing with IOExceptions.
* [Core] Fixed a typo in the global logger class (BREAKING CHANGE! Please find new example in Wiki).
* [Core] Added support for project compilation under macOS (Thanks to @FourOnes).
* [Client] The certificate for encryption is now loaded as a _X509Certificate2_ which prodives more data (Thanks to @cvellan).
* [Client] Fixed an issue in _ManagedClient_ which can cause the client to stop when publishing subscriptions.
* [Client] Fixed an issue in _ManagedClient_ which prevents changing the QoS of an existing subscription (BREAKING CHANGE!).
* [Client] Fixed an issue in _MqttClientOptionsBuilder_ which prevents adding TLS options to the client options when calling _Build()_. (Thanks to @cvellan).
* [Client] Added an overload for subscribing at both clients which reduces required code.
* [Client] Added the _ClientId_ to the application message interceptor context.
* [Server] The application message interceptor can now delete any received application message.
* [Server] Added a ConnectionValidator context to align with other APIs (BREAKING CHANGE! Please find new example in Wiki).
* [Server] Added an interface for the _MqttServerOptions_.
* [Server] Added packet statistics for the connected clients.
* [Server] Fixed a security issue which sends retained packages to a failed subscription.
* [Server] Fixed the response (MaximumQoS) of a subscription (Thanks to @redbeans2017).
* [Server] The keep alive timeouts are now checked for every client (Thanks to @RainerMueller82).
* [Server] The used IP addresses can be specified now (IPAddress.Any as default).
<releaseNotes>* [Core] Add several new extension methods.
* [Client] Fixed an issue in _ManagedMqttClientOptionsBuilder_ when using _WithClientOptions_ and an options builder.
* [Client] Added the "IsStarted" property for the managed client.
* [Client] Optimized stream buffer for UWP apps.
* [Client] Added the _BufferSize_ to the TCP options.
* [Client] Fixed a race condition which leads to exceptions when reconnecting rapidly.
* [Server] Fixed a race condition which leads to exceptions when clients are reconnecting rapidly.
* [Core] Fixed some issues in stream and socket handling.
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2017</copyright>
<copyright>Copyright Christian Kratky 2016-2018</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>
@@ -75,7 +63,6 @@
<!-- .NET Framework -->
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net452\MQTTnet.*" target="lib\net452\"/>
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net461\MQTTnet.*" target="lib\net461\"/>
</files>
+10 -8
View File
@@ -1,6 +1,7 @@
param([string]$assemblyVersion, [string]$nugetVersion)
if ([string]::IsNullOrEmpty($version)) {$version = "0.0.1"}
if ([string]::IsNullOrEmpty($assemblyVersion)) {$assemblyVersion = "0.0.1"}
if ([string]::IsNullOrEmpty($nugetVersion)) {$nugetVersion = "0.0.1"}
$vswhere = ${Env:\ProgramFiles(x86)} + '\Microsoft Visual Studio\Installer\vswhere'
@@ -9,14 +10,14 @@ if ($path) {
$msbuild = join-path $path 'MSBuild\15.0\Bin\MSBuild.exe'
# Build the core library
&$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
&$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
&$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
&$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
&$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
&$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.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 ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.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 ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.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 ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.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 ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.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 ..\Frameworks\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
&$msbuild ..\Frameworks\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"
# Build the RPC extension
&$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
@@ -25,7 +26,8 @@ if ($path) {
&$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
&$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
Remove-Item .\NuGet -Force -Recurse
Remove-Item .\NuGet -Force -Recurse -ErrorAction SilentlyContinue
New-Item -ItemType Directory -Force -Path .\NuGet
.\NuGet.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
.\NuGet.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
Binary file not shown.
+9
View File
@@ -0,0 +1,9 @@
# Import codeSigningKey.pfx
In order to import the key for code signing on a new developer machine use the following command within the VS developer command line:
> sn –i codeSigningKey.pfx VS_KEY_EFCA4C5B6DFD4B4F
The container name may be different.
# Check if the assembly has a strong name
> sn -vf MQTTnet.dll
@@ -20,6 +20,7 @@
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
<DefaultLanguage>en</DefaultLanguage>
<LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
</PropertyGroup>
@@ -16,9 +16,9 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.0" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.0" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.0" />
<PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.1" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.1" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.1" />
</ItemGroup>
<ItemGroup>
@@ -7,7 +7,7 @@ using MQTTnet.Serializer;
namespace MQTTnet.Adapter
{
public interface IMqttChannelAdapter
public interface IMqttChannelAdapter : IDisposable
{
IMqttPacketSerializer PacketSerializer { get; }
@@ -14,12 +14,12 @@ using MQTTnet.Serializer;
namespace MQTTnet.Adapter
{
public class MqttChannelAdapter : IMqttChannelAdapter
public sealed class MqttChannelAdapter : IMqttChannelAdapter
{
private const uint ErrorOperationAborted = 0x800703E3;
private const int ReadBufferSize = 4096; // TODO: Move buffer size to config
private static readonly byte[] EmptyBody = new byte[0];
private bool _isDisposed;
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly IMqttNetLogger _logger;
private readonly IMqttChannel _channel;
@@ -35,6 +35,7 @@ namespace MQTTnet.Adapter
public Task ConnectAsync(TimeSpan timeout)
{
ThrowIfDisposed();
_logger.Trace<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
return ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
@@ -42,6 +43,7 @@ namespace MQTTnet.Adapter
public Task DisconnectAsync(TimeSpan timeout)
{
ThrowIfDisposed();
_logger.Trace<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
return ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
@@ -49,6 +51,8 @@ namespace MQTTnet.Adapter
public Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets)
{
ThrowIfDisposed();
return ExecuteAndWrapExceptionAsync(async () =>
{
await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
@@ -56,6 +60,11 @@ namespace MQTTnet.Adapter
{
foreach (var packet in packets)
{
if (cancellationToken.IsCancellationRequested)
{
return;
}
if (packet == null)
{
continue;
@@ -66,10 +75,20 @@ namespace MQTTnet.Adapter
var chunks = PacketSerializer.Serialize(packet);
foreach (var chunk in chunks)
{
if (cancellationToken.IsCancellationRequested)
{
return;
}
await _channel.SendStream.WriteAsync(chunk.Array, chunk.Offset, chunk.Count, cancellationToken).ConfigureAwait(false);
}
}
if (cancellationToken.IsCancellationRequested)
{
return;
}
if (timeout > TimeSpan.Zero)
{
await _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
@@ -88,31 +107,40 @@ namespace MQTTnet.Adapter
public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
ThrowIfDisposed();
MqttBasePacket packet = null;
await ExecuteAndWrapExceptionAsync(async () =>
{
ReceivedMqttPacket receivedMqttPacket;
if (timeout > TimeSpan.Zero)
ReceivedMqttPacket receivedMqttPacket = null;
try
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
}
else
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
}
if (timeout > TimeSpan.Zero)
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
}
else
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
}
if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
{
throw new TaskCanceledException();
}
if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
{
throw new TaskCanceledException();
}
packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
if (packet == null)
{
throw new MqttProtocolViolationException("Received malformed packet.");
}
packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
if (packet == null)
{
throw new MqttProtocolViolationException("Received malformed packet.");
}
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
}
finally
{
receivedMqttPacket?.Dispose();
}
}).ConfigureAwait(false);
return packet;
@@ -120,7 +148,7 @@ namespace MQTTnet.Adapter
private static async Task<ReceivedMqttPacket> ReceiveAsync(Stream stream, CancellationToken cancellationToken)
{
var header = MqttPacketReader.ReadHeaderFromSource(stream, cancellationToken);
var header = await MqttPacketReader.ReadHeaderAsync(stream, cancellationToken).ConfigureAwait(false);
if (header == null)
{
return null;
@@ -128,18 +156,35 @@ namespace MQTTnet.Adapter
if (header.BodyLength == 0)
{
return new ReceivedMqttPacket(header, EmptyBody);
return new ReceivedMqttPacket(header, new MemoryStream(new byte[0], false));
}
var body = new byte[header.BodyLength];
var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream();
var offset = 0;
do
var buffer = new byte[ReadBufferSize];
while (body.Length < header.BodyLength)
{
var readBytesCount = await stream.ReadAsync(body, offset, body.Length - offset, cancellationToken).ConfigureAwait(false);
offset += readBytesCount;
} while (offset < header.BodyLength);
var bytesLeft = header.BodyLength - (int)body.Length;
if (bytesLeft > buffer.Length)
{
bytesLeft = buffer.Length;
}
var readBytesCount = await stream.ReadAsync(buffer, 0, bytesLeft, cancellationToken).ConfigureAwait(false);
// Check if the client closed the connection before sending the full body.
if (readBytesCount == 0)
{
throw new MqttCommunicationException("Connection closed while reading remaining packet body.");
}
// Here is no need to await because internally only an array is used and no real I/O operation is made.
// Using async here will only generate overhead.
body.Write(buffer, 0, readBytesCount);
}
body.Seek(0L, SeekOrigin.Begin);
return new ReceivedMqttPacket(header, body);
}
@@ -167,7 +212,7 @@ namespace MQTTnet.Adapter
}
catch (COMException comException)
{
if ((uint) comException.HResult == ErrorOperationAborted)
if ((uint)comException.HResult == ErrorOperationAborted)
{
throw new OperationCanceledException();
}
@@ -191,5 +236,20 @@ namespace MQTTnet.Adapter
throw new MqttCommunicationException(exception);
}
}
public void Dispose()
{
_isDisposed = true;
_semaphore?.Dispose();
_channel?.Dispose();
}
private void ThrowIfDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(MqttChannelAdapter));
}
}
}
}
@@ -1,17 +0,0 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Packets;
namespace MQTTnet.Adapter
{
public static class MqttChannelAdapterExtensions
{
public static Task SendPacketsAsync(this IMqttChannelAdapter adapter, TimeSpan timeout, CancellationToken cancellationToken, params MqttBasePacket[] packets)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
return adapter.SendPacketsAsync(timeout, cancellationToken, packets);
}
}
}
@@ -1,11 +1,12 @@
using System;
using System.IO;
using MQTTnet.Packets;
namespace MQTTnet.Adapter
{
public class ReceivedMqttPacket
public sealed class ReceivedMqttPacket : IDisposable
{
public ReceivedMqttPacket(MqttPacketHeader header, byte[] body)
public ReceivedMqttPacket(MqttPacketHeader header, MemoryStream body)
{
Header = header ?? throw new ArgumentNullException(nameof(header));
Body = body ?? throw new ArgumentNullException(nameof(body));
@@ -13,6 +14,11 @@ namespace MQTTnet.Adapter
public MqttPacketHeader Header { get; }
public byte[] Body { get; }
public MemoryStream Body { get; }
public void Dispose()
{
Body?.Dispose();
}
}
}
@@ -1,16 +1,41 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Protocol;
namespace MQTTnet
{
public static class ApplicationMessagePublisherExtensions
{
public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages)
public static Task PublishAsync(this IApplicationMessagePublisher publisher, params MqttApplicationMessage[] applicationMessages)
{
if (client == null) throw new ArgumentNullException(nameof(client));
if (publisher == null) throw new ArgumentNullException(nameof(publisher));
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
return client.PublishAsync(applicationMessages);
return publisher.PublishAsync(applicationMessages);
}
public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic)
{
if (publisher == null) throw new ArgumentNullException(nameof(publisher));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).Build());
}
public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload)
{
if (publisher == null) throw new ArgumentNullException(nameof(publisher));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(payload).Build());
}
public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
{
if (publisher == null) throw new ArgumentNullException(nameof(publisher));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(payload).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
}
}
}
@@ -1,9 +1,10 @@
using System.IO;
using System;
using System.IO;
using System.Threading.Tasks;
namespace MQTTnet.Channel
{
public interface IMqttChannel
public interface IMqttChannel : IDisposable
{
Stream SendStream { get; }
Stream ReceiveStream { get; }
@@ -4,7 +4,7 @@ using System.Threading.Tasks;
namespace MQTTnet.Client
{
public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
{
bool IsConnected { get; }
@@ -13,6 +13,8 @@ namespace MQTTnet.Client
TimeSpan CommunicationTimeout { get; }
TimeSpan KeepAlivePeriod { get; }
TimeSpan? KeepAliveSendInterval { get; set; }
MqttProtocolVersion ProtocolVersion { get; }
IMqttClientChannelOptions ChannelOptions { get; }
@@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@@ -12,23 +13,27 @@ using MQTTnet.Protocol;
namespace MQTTnet.Client
{
public class MqttClient : IMqttClient, IDisposable
public class MqttClient : IMqttClient
{
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
private readonly Stopwatch _sendTracker = new Stopwatch();
private readonly SemaphoreSlim _disconnectLock = new SemaphoreSlim(1, 1);
private readonly IMqttClientAdapterFactory _adapterFactory;
private readonly MqttPacketDispatcher _packetDispatcher;
private readonly IMqttNetLogger _logger;
private IMqttClientOptions _options;
private bool _isReceivingPackets;
private int _latestPacketIdentifier;
private CancellationTokenSource _cancellationTokenSource;
private Task _packetReceiverTask;
private Task _keepAliveMessageSenderTask;
private IMqttChannelAdapter _adapter;
public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
{
_adapterFactory = channelFactory ?? throw new ArgumentNullException(nameof(channelFactory));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_packetDispatcher = new MqttPacketDispatcher(logger);
}
@@ -49,23 +54,25 @@ namespace MQTTnet.Client
{
_options = options;
_cancellationTokenSource = new CancellationTokenSource();
_latestPacketIdentifier = 0;
_packetIdentifierProvider.Reset();
_packetDispatcher.Reset();
_adapter = _adapterFactory.CreateClientAdapter(options.ChannelOptions, _logger);
_logger.Trace<MqttClient>("Trying to connect with server.");
await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
_logger.Trace<MqttClient>("Connection with server established.");
await StartReceivingPacketsAsync(_cancellationTokenSource.Token).ConfigureAwait(false);
var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
await StartReceivingPacketsAsync().ConfigureAwait(false);
var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
_logger.Trace<MqttClient>("MQTT connection with server established.");
_sendTracker.Restart();
if (_options.KeepAlivePeriod != TimeSpan.Zero)
{
StartSendingKeepAliveMessages(_cancellationTokenSource.Token);
StartSendingKeepAliveMessages();
}
IsConnected = true;
@@ -75,7 +82,8 @@ namespace MQTTnet.Client
catch (Exception exception)
{
_logger.Error<MqttClient>(exception, "Error while connecting with server.");
await DisconnectInternalAsync().ConfigureAwait(false);
await DisconnectInternalAsync(exception).ConfigureAwait(false);
throw;
}
}
@@ -89,11 +97,14 @@ namespace MQTTnet.Client
try
{
await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false);
if (!_cancellationTokenSource.IsCancellationRequested)
{
await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false);
}
}
finally
{
await DisconnectInternalAsync().ConfigureAwait(false);
await DisconnectInternalAsync(null).ConfigureAwait(false);
}
}
@@ -105,7 +116,7 @@ namespace MQTTnet.Client
var subscribePacket = new MqttSubscribePacket
{
PacketIdentifier = GetNewPacketIdentifier(),
PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(),
TopicFilters = topicFilters.ToList()
};
@@ -127,7 +138,7 @@ namespace MQTTnet.Client
var unsubscribePacket = new MqttUnsubscribePacket
{
PacketIdentifier = GetNewPacketIdentifier(),
PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(),
TopicFilters = topicFilters.ToList()
};
@@ -139,7 +150,7 @@ namespace MQTTnet.Client
ThrowIfNotConnected();
var publishPackets = applicationMessages.Select(m => m.ToPublishPacket());
var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key);
var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key);
foreach (var qosGroup in packetGroups)
{
@@ -148,14 +159,14 @@ namespace MQTTnet.Client
case MqttQualityOfServiceLevel.AtMostOnce:
{
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosGroup).ConfigureAwait(false);
await SendAsync((MqttPublishPacket[])qosGroup.ToArray()).ConfigureAwait(false);
break;
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
foreach (var publishPacket in qosGroup)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
}
@@ -165,9 +176,15 @@ namespace MQTTnet.Client
{
foreach (var publishPacket in qosGroup)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
var pubRelPacket = new MqttPubRelPacket
{
PacketIdentifier = pubRecPacket.PacketIdentifier
};
await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket).ConfigureAwait(false);
}
break;
@@ -183,6 +200,9 @@ namespace MQTTnet.Client
public void Dispose()
{
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_adapter?.Dispose();
}
private async Task<MqttConnAckPacket> AuthenticateAsync(MqttApplicationMessage willApplicationMessage)
@@ -216,34 +236,50 @@ namespace MQTTnet.Client
if (IsConnected) throw new MqttProtocolViolationException(message);
}
private async Task DisconnectInternalAsync()
private async Task DisconnectInternalAsync(Exception exception)
{
await _disconnectLock.WaitAsync();
var clientWasConnected = IsConnected;
IsConnected = false;
var cts = _cancellationTokenSource;
if (cts == null || cts.IsCancellationRequested)
{
return;
}
cts.Cancel(false);
cts.Dispose();
_cancellationTokenSource = null;
try
{
IsConnected = false;
if (_cancellationTokenSource == null || _cancellationTokenSource.IsCancellationRequested)
{
return;
}
_cancellationTokenSource.Cancel(false);
if (_packetReceiverTask != null)
{
Task.WaitAll(_packetReceiverTask);
}
if (_keepAliveMessageSenderTask != null)
{
Task.WaitAll(_keepAliveMessageSenderTask);
}
await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
_logger.Info<MqttClient>("Disconnected from adapter.");
_logger.Trace<MqttClient>("Disconnected from adapter.");
}
catch (Exception exception)
catch (Exception adapterException)
{
_logger.Warning<MqttClient>(exception, "Error while disconnecting from adapter.");
_logger.Warning<MqttClient>(adapterException, "Error while disconnecting from adapter.");
}
finally
{
_adapter?.Dispose();
_adapter = null;
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_disconnectLock.Release();
_logger.Info<MqttClient>("Disconnected.");
Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected));
Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected, exception));
}
}
@@ -252,7 +288,7 @@ namespace MQTTnet.Client
try
{
_logger.Info<MqttClient>("Received <<< {0}", packet);
if (packet is MqttPublishPacket publishPacket)
{
await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false);
@@ -300,6 +336,11 @@ namespace MQTTnet.Client
private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket)
{
if (_cancellationTokenSource.IsCancellationRequested)
{
return Task.FromResult(0);
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{
FireApplicationMessageReceivedEvent(publishPacket);
@@ -324,61 +365,77 @@ namespace MQTTnet.Client
private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)
{
return SendAsync(pubRelPacket.CreateResponse<MqttPubCompPacket>());
var response = new MqttPubCompPacket
{
PacketIdentifier = pubRelPacket.PacketIdentifier
};
return SendAsync(response);
}
private Task SendAsync(MqttBasePacket packet)
private Task SendAsync(params MqttBasePacket[] packets)
{
return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packet);
_sendTracker.Restart();
return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packets);
}
private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
{
var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.CommunicationTimeout);
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false);
ushort? identifier = null;
if (requestPacket is IMqttPacketWithIdentifier requestPacketWithIdentifier)
{
identifier = requestPacketWithIdentifier.PacketIdentifier;
}
var packetAwaiter = _packetDispatcher.WaitForPacketAsync(typeof(TResponsePacket), identifier, _options.CommunicationTimeout);
await SendAsync(requestPacket).ConfigureAwait(false);
return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
}
private ushort GetNewPacketIdentifier()
{
return (ushort)Interlocked.Increment(ref _latestPacketIdentifier);
}
private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
private async Task SendKeepAliveMessagesAsync()
{
_logger.Info<MqttClient>("Start sending keep alive packets.");
try
{
while (!cancellationToken.IsCancellationRequested)
while (!_cancellationTokenSource.Token.IsCancellationRequested)
{
await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false);
var keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75);
if (_options.KeepAliveSendInterval.HasValue)
{
keepAliveSendInterval = _options.KeepAliveSendInterval.Value;
}
if (_sendTracker.Elapsed > keepAliveSendInterval)
{
await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
}
await Task.Delay(keepAliveSendInterval, _cancellationTokenSource.Token).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
if (cancellationToken.IsCancellationRequested)
{
return;
}
await DisconnectInternalAsync().ConfigureAwait(false);
}
catch (MqttCommunicationException exception)
{
if (cancellationToken.IsCancellationRequested)
{
return;
}
_logger.Warning<MqttClient>(exception, "MQTT communication exception while sending/receiving keep alive packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.Warning<MqttClient>(exception, "Unhandled exception while sending/receiving keep alive packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
if (_cancellationTokenSource.Token.IsCancellationRequested)
{
return;
}
if (exception is MqttCommunicationException)
{
_logger.Warning<MqttClient>(exception, "MQTT communication exception while sending/receiving keep alive packets.");
}
else
{
_logger.Warning<MqttClient>(exception, "Unhandled exception while sending/receiving keep alive packets.");
}
await DisconnectInternalAsync(exception).ConfigureAwait(false);
}
finally
{
@@ -386,49 +443,47 @@ namespace MQTTnet.Client
}
}
private async Task ReceivePacketsAsync(CancellationToken cancellationToken)
private async Task ReceivePacketsAsync()
{
_logger.Info<MqttClient>("Start receiving packets.");
try
{
while (!cancellationToken.IsCancellationRequested)
while (!_cancellationTokenSource.Token.IsCancellationRequested)
{
_isReceivingPackets = true;
var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
if (_cancellationTokenSource.Token.IsCancellationRequested)
{
return;
}
StartProcessReceivedPacket(packet, cancellationToken);
StartProcessReceivedPacket(packet);
}
}
catch (OperationCanceledException)
{
if (cancellationToken.IsCancellationRequested)
{
return;
}
await DisconnectInternalAsync().ConfigureAwait(false);
}
catch (MqttCommunicationException exception)
{
if (cancellationToken.IsCancellationRequested)
{
return;
}
_logger.Warning<MqttClient>(exception, "MQTT communication exception while receiving packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.Error<MqttClient>(exception, "Unhandled exception while receiving packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
if (_cancellationTokenSource.IsCancellationRequested)
{
return;
}
if (exception is MqttCommunicationException)
{
_logger.Warning<MqttClient>(exception, "MQTT communication exception while receiving packets.");
}
else
{
_logger.Error<MqttClient>(exception, "Unhandled exception while receiving packets.");
}
await DisconnectInternalAsync(exception).ConfigureAwait(false);
}
finally
{
@@ -436,38 +491,26 @@ namespace MQTTnet.Client
}
}
private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken)
private void StartProcessReceivedPacket(MqttBasePacket packet)
{
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Run(
async () => await ProcessReceivedPacketAsync(packet).ConfigureAwait(false),
cancellationToken).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Run(() => ProcessReceivedPacketAsync(packet), _cancellationTokenSource.Token);
}
private async Task StartReceivingPacketsAsync(CancellationToken cancellationToken)
private async Task StartReceivingPacketsAsync()
{
_isReceivingPackets = false;
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Run(
async () => await ReceivePacketsAsync(cancellationToken).ConfigureAwait(false),
cancellationToken).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
_packetReceiverTask = Task.Run(ReceivePacketsAsync, _cancellationTokenSource.Token);
while (!_isReceivingPackets && !cancellationToken.IsCancellationRequested)
while (!_isReceivingPackets && !_cancellationTokenSource.Token.IsCancellationRequested)
{
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken).ConfigureAwait(false);
await Task.Delay(TimeSpan.FromMilliseconds(100), _cancellationTokenSource.Token).ConfigureAwait(false);
}
}
private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
private void StartSendingKeepAliveMessages()
{
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Run(
async () => await SendKeepAliveMessagesAsync(cancellationToken).ConfigureAwait(false),
cancellationToken).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
_keepAliveMessageSenderTask = Task.Run(SendKeepAliveMessagesAsync, _cancellationTokenSource.Token);
}
}
}
@@ -4,11 +4,14 @@ namespace MQTTnet.Client
{
public class MqttClientDisconnectedEventArgs : EventArgs
{
public MqttClientDisconnectedEventArgs(bool clientWasConnected)
public MqttClientDisconnectedEventArgs(bool clientWasConnected, Exception exception)
{
ClientWasConnected = clientWasConnected;
Exception = exception;
}
public bool ClientWasConnected { get; }
public Exception Exception { get; }
}
}
@@ -24,6 +24,14 @@ namespace MQTTnet.Client
return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
}
public static Task<IList<MqttSubscribeResult>> SubscribeAsync(this IMqttClient client, string topic)
{
if (client == null) throw new ArgumentNullException(nameof(client));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
}
public static Task UnsubscribeAsync(this IMqttClient client, params string[] topicFilters)
{
if (client == null) throw new ArgumentNullException(nameof(client));
@@ -13,7 +13,9 @@ namespace MQTTnet.Client
public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials();
public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5);
public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15);
public TimeSpan? KeepAliveSendInterval { get; set; }
public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
@@ -6,6 +6,8 @@
public int? Port { get; set; }
public int BufferSize { get; set; } = 20 * 4096;
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
}
}
@@ -9,7 +9,7 @@ namespace MQTTnet.Client
public IDictionary<string, string> RequestHeaders { get; set; }
public ICollection<string> SubProtocols { get; set; }
public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" };
public CookieContainer CookieContainer { get; set; }
@@ -10,8 +10,7 @@ namespace MQTTnet.Client
{
public class MqttPacketDispatcher
{
private readonly ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>> _packetByResponseType = new ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>>();
private readonly ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>> _packetByResponseTypeAndIdentifier = new ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>>();
private readonly ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>> _awaiters = new ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>>();
private readonly IMqttNetLogger _logger;
public MqttPacketDispatcher(IMqttNetLogger logger)
@@ -19,11 +18,9 @@ namespace MQTTnet.Client
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
public async Task<MqttBasePacket> WaitForPacketAsync(MqttBasePacket request, Type responseType, TimeSpan timeout)
public async Task<MqttBasePacket> WaitForPacketAsync(Type responseType, ushort? identifier, TimeSpan timeout)
{
if (request == null) throw new ArgumentNullException(nameof(request));
var packetAwaiter = AddPacketAwaiter(request, responseType);
var packetAwaiter = AddPacketAwaiter(responseType, identifier);
try
{
return await packetAwaiter.Task.TimeoutAfter(timeout).ConfigureAwait(false);
@@ -35,7 +32,7 @@ namespace MQTTnet.Client
}
finally
{
RemovePacketAwaiter(request, responseType);
RemovePacketAwaiter(responseType, identifier);
}
}
@@ -44,21 +41,20 @@ namespace MQTTnet.Client
if (packet == null) throw new ArgumentNullException(nameof(packet));
var type = packet.GetType();
if (packet is IMqttPacketWithIdentifier withIdentifier)
if (_awaiters.TryGetValue(type, out var byId))
{
if (_packetByResponseTypeAndIdentifier.TryGetValue(type, out var byid))
ushort? identifier = 0;
if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
{
if (byid.TryRemove(withIdentifier.PacketIdentifier, out var tcs))
{
tcs.TrySetResult(packet);
return;
}
identifier = packetWithIdentifier.PacketIdentifier;
}
if (byId.TryRemove(identifier.Value, out var tcs))
{
tcs.TrySetResult(packet);
return;
}
}
else if (_packetByResponseType.TryRemove(type, out var tcs))
{
tcs.TrySetResult(packet);
return;
}
throw new InvalidOperationException($"Packet of type '{type.Name}' not handled or dispatched.");
@@ -66,38 +62,36 @@ namespace MQTTnet.Client
public void Reset()
{
_packetByResponseTypeAndIdentifier.Clear();
_packetByResponseType.Clear();
_awaiters.Clear();
}
private TaskCompletionSource<MqttBasePacket> AddPacketAwaiter(MqttBasePacket request, Type responseType)
private TaskCompletionSource<MqttBasePacket> AddPacketAwaiter(Type responseType, ushort? identifier)
{
var tcs = new TaskCompletionSource<MqttBasePacket>();
if (request is IMqttPacketWithIdentifier requestWithIdentifier)
if (!identifier.HasValue)
{
var byId = _packetByResponseTypeAndIdentifier.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
byId[requestWithIdentifier.PacketIdentifier] = tcs;
identifier = 0;
}
else
var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
if (!byId.TryAdd(identifier.Value, tcs))
{
_packetByResponseType[responseType] = tcs;
throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{responseType}' with identifier {identifier}.");
}
return tcs;
}
private void RemovePacketAwaiter(MqttBasePacket request, Type responseType)
private void RemovePacketAwaiter(Type responseType, ushort? identifier)
{
if (request is IMqttPacketWithIdentifier requestWithIdentifier)
if (!identifier.HasValue)
{
var byId = _packetByResponseTypeAndIdentifier.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
byId.TryRemove(requestWithIdentifier.PacketIdentifier, out var _);
}
else
{
_packetByResponseType.TryRemove(responseType, out var _);
identifier = 0;
}
var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
byId.TryRemove(identifier.Value, out var _);
}
}
}
@@ -0,0 +1,32 @@
namespace MQTTnet.Client
{
public class MqttPacketIdentifierProvider
{
private readonly object _syncRoot = new object();
private ushort _value;
public void Reset()
{
lock (_syncRoot)
{
_value = 0;
}
}
public ushort GetNewPacketIdentifier()
{
lock (_syncRoot)
{
_value++;
if (_value == 0)
{
// As per official MQTT documentation the package identifier should never be 0.
_value = 1;
}
return _value;
}
}
}
}
@@ -28,5 +28,16 @@ namespace MQTTnet.Diagnostics
public string Message { get; }
public Exception Exception { get; }
public override string ToString()
{
var result = $"[{Timestamp:O}] [{LogId}] [{ThreadId}] [{Source}] [{Level}]: {Message}";
if (Exception != null)
{
result += Environment.NewLine + Exception;
}
return result;
}
}
}
@@ -110,22 +110,14 @@ namespace MQTTnet.Implementations
{
// It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
}
catch (SocketException exception)
catch (Exception exception)
{
if (exception.SocketErrorCode == SocketError.OperationAborted)
if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
{
return;
}
_logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
}
catch (Exception exception)
{
_logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
}
finally
{
//excessive CPU consumed if in endless loop of socket errors
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
}
}
@@ -153,22 +145,14 @@ namespace MQTTnet.Implementations
{
// It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
}
catch (SocketException exception)
catch (Exception exception)
{
if (exception.SocketErrorCode == SocketError.OperationAborted)
if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
{
return;
}
_logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
}
catch (Exception exception)
{
_logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at TLS endpoint.");
}
finally
{
//excessive CPU consumed if in endless loop of socket errors
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
}
}
@@ -13,14 +13,21 @@ using MQTTnet.Client;
namespace MQTTnet.Implementations
{
public sealed class MqttTcpChannel : IMqttChannel, IDisposable
public sealed class MqttTcpChannel : IMqttChannel
{
// ReSharper disable once MemberCanBePrivate.Global
// ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user.
private readonly int _bufferSize = BufferSize;
private readonly MqttClientTcpOptions _options;
private StreamSocket _socket;
public MqttTcpChannel(MqttClientTcpOptions options)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_bufferSize = _options.BufferSize;
}
public MqttTcpChannel(StreamSocket socket)
@@ -69,20 +76,56 @@ namespace MQTTnet.Implementations
public void Dispose()
{
SendStream?.Dispose();
SendStream = null;
try
{
SendStream?.Dispose();
}
catch (ObjectDisposedException)
{
}
catch (NullReferenceException)
{
}
finally
{
SendStream = null;
}
ReceiveStream?.Dispose();
ReceiveStream = null;
try
{
ReceiveStream?.Dispose();
}
catch (ObjectDisposedException)
{
}
catch (NullReferenceException)
{
}
finally
{
ReceiveStream = null;
}
_socket?.Dispose();
_socket = null;
try
{
_socket?.Dispose();
}
catch (ObjectDisposedException)
{
}
catch (NullReferenceException)
{
}
finally
{
_socket = null;
}
}
private void CreateStreams()
{
SendStream = _socket.OutputStream.AsStreamForWrite();
ReceiveStream = _socket.InputStream.AsStreamForRead();
SendStream = _socket.OutputStream.AsStreamForWrite(_bufferSize);
ReceiveStream = _socket.InputStream.AsStreamForRead(_bufferSize);
}
private static Certificate LoadCertificate(MqttClientTcpOptions options)
@@ -12,17 +12,20 @@ using MQTTnet.Client;
namespace MQTTnet.Implementations
{
public sealed class MqttTcpChannel : IMqttChannel, IDisposable
public sealed class MqttTcpChannel : IMqttChannel
{
private readonly MqttClientTcpOptions _options;
//todo: this can be used with min dependency NetStandard1.6
#if NET452 || NET461
// ReSharper disable once MemberCanBePrivate.Global
// ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user.
private readonly int _bufferSize = BufferSize;
#else
private readonly int _bufferSize = 0;
#endif
private readonly MqttClientTcpOptions _options;
private Socket _socket;
private SslStream _sslStream;
@@ -32,6 +35,7 @@ namespace MQTTnet.Implementations
public MqttTcpChannel(MqttClientTcpOptions options)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_bufferSize = options.BufferSize;
}
/// <summary>
@@ -43,7 +47,7 @@ namespace MQTTnet.Implementations
_socket = socket ?? throw new ArgumentNullException(nameof(socket));
_sslStream = sslStream;
CreateStreams(socket, sslStream);
CreateStreams();
}
public Stream SendStream { get; private set; }
@@ -58,7 +62,6 @@ namespace MQTTnet.Implementations
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
}
//todo: else brach can be used with min dependency NET46
#if NET452 || NET461
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
#else
@@ -68,10 +71,10 @@ namespace MQTTnet.Implementations
if (_options.TlsOptions.UseTls)
{
_sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(_options), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
}
CreateStreams(_socket, _sslStream);
CreateStreams();
}
public Task DisconnectAsync()
@@ -82,11 +85,70 @@ namespace MQTTnet.Implementations
public void Dispose()
{
_socket?.Dispose();
_socket = null;
var oneStreamIsUsed = SendStream != null && ReceiveStream != null && ReferenceEquals(SendStream, ReceiveStream);
_sslStream?.Dispose();
_sslStream = null;
try
{
SendStream?.Dispose();
}
catch (ObjectDisposedException)
{
}
catch (NullReferenceException)
{
}
finally
{
SendStream = null;
}
try
{
if (!oneStreamIsUsed)
{
ReceiveStream?.Dispose();
}
}
catch (ObjectDisposedException)
{
}
catch (NullReferenceException)
{
}
finally
{
ReceiveStream = null;
}
try
{
_sslStream?.Dispose();
}
catch (ObjectDisposedException)
{
}
catch (NullReferenceException)
{
}
finally
{
_sslStream = null;
}
try
{
_socket?.Dispose();
}
catch (ObjectDisposedException)
{
}
catch (NullReferenceException)
{
}
finally
{
_socket = null;
}
}
private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
@@ -120,15 +182,15 @@ namespace MQTTnet.Implementations
return _options.TlsOptions.AllowUntrustedCertificates;
}
private static X509CertificateCollection LoadCertificates(MqttClientTcpOptions options)
private X509CertificateCollection LoadCertificates()
{
var certificates = new X509CertificateCollection();
if (options.TlsOptions.Certificates == null)
if (_options.TlsOptions.Certificates == null)
{
return certificates;
}
foreach (var certificate in options.TlsOptions.Certificates)
foreach (var certificate in _options.TlsOptions.Certificates)
{
certificates.Add(new X509Certificate2(certificate));
}
@@ -136,20 +198,26 @@ namespace MQTTnet.Implementations
return certificates;
}
private void CreateStreams(Socket socket, Stream sslStream)
private void CreateStreams()
{
var stream = sslStream ?? new NetworkStream(socket);
Stream stream;
if (_sslStream != null)
{
stream = _sslStream;
}
else
{
stream = new NetworkStream(_socket, true);
}
//todo: if branch can be used with min dependency NetStandard1.6
#if NET452 || NET461
SendStream = new BufferedStream(stream, BufferSize);
ReceiveStream = new BufferedStream(stream, BufferSize);
SendStream = new BufferedStream(stream, _bufferSize);
ReceiveStream = new BufferedStream(stream, _bufferSize);
#else
SendStream = stream;
ReceiveStream = stream;
#endif
}
}
}
#endif
@@ -96,8 +96,17 @@ namespace MQTTnet.Implementations
public void Dispose()
{
_webSocket?.Dispose();
_webSocket = null;
try
{
_webSocket?.Dispose();
}
catch (ObjectDisposedException)
{
}
finally
{
_webSocket = null;
}
}
}
}
@@ -11,9 +11,9 @@ namespace MQTTnet.Implementations
{
public class WebSocketStream : Stream
{
private readonly WebSocket _webSocket;
private readonly byte[] _chunkBuffer = new byte[MqttWebSocketChannel.BufferSize];
private readonly byte[] _chunckBuffer = new byte[MqttWebSocketChannel.BufferSize];
private readonly Queue<byte> _buffer = new Queue<byte>(MqttWebSocketChannel.BufferSize);
private readonly WebSocket _webSocket;
public WebSocketStream(WebSocket webSocket)
{
@@ -55,25 +55,28 @@ namespace MQTTnet.Implementations
// Use existing date from buffer.
while (count > 0 && _buffer.Any())
{
buffer[offset++] = _buffer.Dequeue();
buffer[offset] = _buffer.Dequeue();
count--;
bytesRead++;
offset++;
}
if (count == 0)
{
return bytesRead;
}
// Fetch new data if the buffer is not full.
while (_webSocket.State == WebSocketState.Open)
{
await FetchChunkAsync(cancellationToken);
await FetchChunkAsync(cancellationToken).ConfigureAwait(false);
while (count > 0 && _buffer.Any())
{
buffer[offset++] = _buffer.Dequeue();
buffer[offset] = _buffer.Dequeue();
count--;
bytesRead++;
offset++;
}
if (count == 0)
@@ -111,19 +114,24 @@ namespace MQTTnet.Implementations
}
private async Task FetchChunkAsync(CancellationToken cancellationToken)
{
var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(_chunkBuffer, 0, _chunkBuffer.Length), cancellationToken).ConfigureAwait(false);
for (var i = 0; i < response.Count; i++)
{
var @byte = _chunkBuffer[i];
_buffer.Enqueue(@byte);
}
{
var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(_chunckBuffer, 0, _chunckBuffer.Length), cancellationToken).ConfigureAwait(false);
if (response.MessageType == WebSocketMessageType.Close)
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
}
else if (response.MessageType == WebSocketMessageType.Binary)
{
for (var i = 0; i < response.Count; i++)
{
_buffer.Enqueue(_chunckBuffer[i]);
}
}
else if (response.MessageType == WebSocketMessageType.Text)
{
throw new MqttProtocolViolationException("WebSocket channel received TEXT message.");
}
}
}
}
@@ -30,7 +30,7 @@ namespace MQTTnet.Internal
if (task.IsFaulted)
{
throw new MqttCommunicationException(task.Exception.GetBaseException());
throw new MqttCommunicationException(task.Exception?.GetBaseException());
}
}
finally
@@ -15,8 +15,7 @@
<Description />
<Authors />
<PackageId />
<SignAssembly>true</SignAssembly>
<AssemblyOriginatorKeyFile>codeSigningKey.pfx</AssemblyOriginatorKeyFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
</PropertyGroup>
@@ -29,6 +28,7 @@
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
<DefaultLanguage>en</DefaultLanguage>
<LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
</PropertyGroup>
@@ -0,0 +1,19 @@
using System;
namespace MQTTnet.ManagedClient
{
public class ApplicationMessageProcessedEventArgs : EventArgs
{
public ApplicationMessageProcessedEventArgs(MqttApplicationMessage applicationMessage, Exception exception)
{
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
Exception = exception;
}
public MqttApplicationMessage ApplicationMessage { get; }
public Exception Exception { get; }
public bool HasFailed => Exception != null;
public bool HasSucceeded => Exception == null;
}
}
@@ -5,13 +5,16 @@ using MQTTnet.Client;
namespace MQTTnet.ManagedClient
{
public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
{
bool IsStarted { get; }
bool IsConnected { get; }
event EventHandler<MqttClientConnectedEventArgs> Connected;
event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
Task StartAsync(IManagedMqttClientOptions options);
Task StopAsync();
@@ -11,11 +11,12 @@ using MQTTnet.Protocol;
namespace MQTTnet.ManagedClient
{
public class ManagedMqttClient : IManagedMqttClient, IDisposable
public class ManagedMqttClient : IManagedMqttClient
{
private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly SemaphoreSlim _subscriptionsSemaphore = new SemaphoreSlim(1, 1);
private readonly List<string> _unsubscriptions = new List<string>();
private readonly IMqttClient _mqttClient;
private readonly IMqttNetLogger _logger;
@@ -27,7 +28,7 @@ namespace MQTTnet.ManagedClient
private IManagedMqttClientOptions _options;
private bool _subscriptionsNotPushed;
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -39,10 +40,13 @@ namespace MQTTnet.ManagedClient
}
public bool IsConnected => _mqttClient.IsConnected;
public bool IsStarted => _connectionCancellationToken != null;
public event EventHandler<MqttClientConnectedEventArgs> Connected;
public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
public event EventHandler SynchronizingSubscriptionsFailed;
public async Task StartAsync(IManagedMqttClientOptions options)
{
@@ -57,7 +61,7 @@ namespace MQTTnet.ManagedClient
if (_connectionCancellationToken != null) throw new InvalidOperationException("The managed client is already started.");
_options = options;
if (_options.Storage != null)
{
_storageManager = new ManagedMqttClientStorageManager(_options.Storage);
@@ -65,9 +69,9 @@ namespace MQTTnet.ManagedClient
}
_connectionCancellationToken = new CancellationTokenSource();
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Run(async () => await MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token).ConfigureAwait(false);
Task.Run(async () => await MaintainConnectionAsync(_connectionCancellationToken.Token).ConfigureAwait(false), _connectionCancellationToken.Token).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
_logger.Info<ManagedMqttClient>("Started");
@@ -85,7 +89,7 @@ namespace MQTTnet.ManagedClient
return Task.FromResult(0);
}
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
@@ -96,7 +100,7 @@ namespace MQTTnet.ManagedClient
{
await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
}
_messageQueue.Add(applicationMessage);
}
}
@@ -129,6 +133,7 @@ namespace MQTTnet.ManagedClient
{
if (_subscriptions.Remove(topic))
{
_unsubscriptions.Add(topic);
_subscriptionsNotPushed = true;
}
}
@@ -153,7 +158,7 @@ namespace MQTTnet.ManagedClient
{
while (!cancellationToken.IsCancellationRequested)
{
await TryMaintainConnectionAsync(cancellationToken);
await TryMaintainConnectionAsync(cancellationToken).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
@@ -184,11 +189,11 @@ namespace MQTTnet.ManagedClient
if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
{
await PushSubscriptionsAsync().ConfigureAwait(false);
await SynchronizeSubscriptionsAsync().ConfigureAwait(false);
StartPublishing();
return;
}
@@ -209,7 +214,7 @@ namespace MQTTnet.ManagedClient
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while maintaining connection.");
}
}
private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
{
try
@@ -227,7 +232,7 @@ namespace MQTTnet.ManagedClient
continue;
}
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
@@ -245,6 +250,7 @@ namespace MQTTnet.ManagedClient
private async Task TryPublishQueuedMessageAsync(MqttApplicationMessage message)
{
Exception transmitException = null;
try
{
await _mqttClient.PublishAsync(message).ConfigureAwait(false);
@@ -256,6 +262,8 @@ namespace MQTTnet.ManagedClient
}
catch (MqttCommunicationException exception)
{
transmitException = exception;
_logger.Warning<ManagedMqttClient>(exception, "Publishing application message failed.");
if (message.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
@@ -265,39 +273,60 @@ namespace MQTTnet.ManagedClient
}
catch (Exception exception)
{
transmitException = exception;
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while publishing queued application message.");
}
finally
{
ApplicationMessageProcessed?.Invoke(this, new ApplicationMessageProcessedEventArgs(message, transmitException));
}
}
private async Task PushSubscriptionsAsync()
private async Task SynchronizeSubscriptionsAsync()
{
_logger.Info<ManagedMqttClient>(nameof(ManagedMqttClient), "Synchronizing subscriptions");
List<TopicFilter> subscriptions;
List<string> unsubscriptions;
await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
try
{
subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList();
unsubscriptions = new List<string>(_unsubscriptions);
_unsubscriptions.Clear();
_subscriptionsNotPushed = false;
}
finally
{
_subscriptionsSemaphore.Release();
}
if (!subscriptions.Any())
if (!subscriptions.Any() && !unsubscriptions.Any())
{
return;
}
try
{
await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false);
if (subscriptions.Any())
{
await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false);
}
if (unsubscriptions.Any())
{
await _mqttClient.UnsubscribeAsync(unsubscriptions).ConfigureAwait(false);
}
}
catch (Exception exception)
{
_logger.Warning<ManagedMqttClient>(exception, "Synchronizing subscriptions failed");
_logger.Warning<ManagedMqttClient>(exception, "Synchronizing subscriptions failed.");
_subscriptionsNotPushed = true;
SynchronizingSubscriptionsFailed?.Invoke(this, EventArgs.Empty);
}
}
@@ -21,6 +21,14 @@ namespace MQTTnet.ManagedClient
return managedClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
}
public static Task SubscribeAsync(this IManagedMqttClient managedClient, string topic)
{
if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return managedClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
}
public static Task UnsubscribeAsync(this IManagedMqttClient managedClient, params string[] topicFilters)
{
if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
@@ -6,6 +6,7 @@ namespace MQTTnet.ManagedClient
public class ManagedMqttClientOptionsBuilder
{
private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
private MqttClientOptionsBuilder _clientOptionsBuilder;
public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value)
{
@@ -21,7 +22,24 @@ namespace MQTTnet.ManagedClient
public ManagedMqttClientOptionsBuilder WithClientOptions(IMqttClientOptions value)
{
if (_clientOptionsBuilder != null)
{
throw new InvalidOperationException("Cannot use client options builder and client options at the same time.");
}
_options.ClientOptions = value ?? throw new ArgumentNullException(nameof(value));
return this;
}
public ManagedMqttClientOptionsBuilder WithClientOptions(MqttClientOptionsBuilder builder)
{
if (_options.ClientOptions != null)
{
throw new InvalidOperationException("Cannot use client options builder and client options at the same time.");
}
_clientOptionsBuilder = builder;
return this;
}
@@ -29,15 +47,22 @@ namespace MQTTnet.ManagedClient
{
if (options == null) throw new ArgumentNullException(nameof(options));
var builder = new MqttClientOptionsBuilder();
options(builder);
_options.ClientOptions = builder.Build();
if (_clientOptionsBuilder == null)
{
_clientOptionsBuilder = new MqttClientOptionsBuilder();
}
options(_clientOptionsBuilder);
return this;
}
public ManagedMqttClientOptions Build()
{
if (_clientOptionsBuilder != null)
{
_options.ClientOptions = _clientOptionsBuilder.Build();
}
if (_options.ClientOptions == null)
{
throw new InvalidOperationException("The ClientOptions cannot be null.");
@@ -0,0 +1,25 @@
using System;
using System.Text;
namespace MQTTnet
{
public static class MqttApplicationMessageExtensions
{
public static string ConvertPayloadToString(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
if (applicationMessage.Payload == null)
{
return null;
}
if (applicationMessage.Payload.Length == 0)
{
return string.Empty;
}
return Encoding.UTF8.GetString(applicationMessage.Payload, 0, applicationMessage.Payload.Length);
}
}
}
@@ -2,6 +2,6 @@
{
public interface IMqttPacketWithIdentifier
{
ushort PacketIdentifier { get; set; }
ushort? PacketIdentifier { get; set; }
}
}
@@ -2,6 +2,6 @@
{
public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier
{
public ushort PacketIdentifier { get; set; }
public ushort? PacketIdentifier { get; set; }
}
}
@@ -1,26 +0,0 @@
using System;
namespace MQTTnet.Packets
{
public static class MqttPacketExtensions
{
public static TResponsePacket CreateResponse<TResponsePacket>(this MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
var responsePacket = Activator.CreateInstance<TResponsePacket>();
if (responsePacket is IMqttPacketWithIdentifier responsePacketWithIdentifier)
{
if (!(packet is IMqttPacketWithIdentifier requestPacketWithIdentifier))
{
throw new InvalidOperationException("Response packet has PacketIdentifier but request packet does not.");
}
responsePacketWithIdentifier.PacketIdentifier = requestPacketWithIdentifier.PacketIdentifier;
}
return responsePacket;
}
}
}
@@ -6,7 +6,7 @@ namespace MQTTnet.Packets
{
public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
{
public ushort PacketIdentifier { get; set; }
public ushort? PacketIdentifier { get; set; }
public IList<MqttSubscribeReturnCode> SubscribeReturnCodes { get; } = new List<MqttSubscribeReturnCode>();
@@ -5,7 +5,7 @@ namespace MQTTnet.Packets
{
public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
{
public ushort PacketIdentifier { get; set; }
public ushort? PacketIdentifier { get; set; }
public IList<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
@@ -2,7 +2,7 @@
{
public sealed class MqttUnsubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
{
public ushort PacketIdentifier { get; set; }
public ushort? PacketIdentifier { get; set; }
public override string ToString()
{
@@ -4,7 +4,7 @@ namespace MQTTnet.Packets
{
public sealed class MqttUnsubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
{
public ushort PacketIdentifier { get; set; }
public ushort? PacketIdentifier { get; set; }
public IList<string> TopicFilters { get; set; } = new List<string>();
@@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.IO;
using MQTTnet.Packets;
namespace MQTTnet.Serializer
@@ -10,6 +11,6 @@ namespace MQTTnet.Serializer
ICollection<ArraySegment<byte>> Serialize(MqttBasePacket mqttPacket);
MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body);
MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body);
}
}
@@ -22,17 +22,26 @@ namespace MQTTnet.Serializer
public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength;
public static MqttPacketHeader ReadHeaderFromSource(Stream stream, CancellationToken cancellationToken)
public static async Task<MqttPacketHeader> ReadHeaderAsync(Stream stream, CancellationToken cancellationToken)
{
var buffer = stream.ReadByte();
if (buffer == -1)
if (cancellationToken.IsCancellationRequested)
{
return null;
}
var fixedHeader = (byte)buffer;
// Wait for the next package which starts with the header. At this point there will probably
// some large delay and thus the thread should be put back to the pool (await). So ReadByte()
// is not an option here.
var buffer = new byte[1];
var readCount = await stream.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
if (readCount <= 0)
{
return null;
}
var fixedHeader = buffer[0];
var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4);
var bodyLength = ReadBodyLengthFromSource(stream, cancellationToken);
var bodyLength = await ReadBodyLengthAsync(stream, cancellationToken).ConfigureAwait(false);
return new MqttPacketHeader
{
@@ -80,13 +89,14 @@ namespace MQTTnet.Serializer
return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
}
private static int ReadBodyLengthFromSource(Stream stream, CancellationToken cancellationToken)
private static async Task<int> ReadBodyLengthAsync(Stream stream, CancellationToken cancellationToken)
{
// Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
// Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
var multiplier = 1;
var value = 0;
byte encodedByte;
var buffer = new byte[1];
var readBytes = new List<byte>();
do
{
@@ -95,21 +105,22 @@ namespace MQTTnet.Serializer
throw new TaskCanceledException();
}
var buffer = stream.ReadByte();
if (buffer == -1)
var readCount = await stream.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
if (readCount <= 0)
{
throw new MqttCommunicationException("Connection closed while reading remaining length data.");
}
encodedByte = (byte)buffer;
encodedByte = buffer[0];
readBytes.Add(encodedByte);
value += (byte)(encodedByte & 127) * multiplier;
multiplier *= 128;
if (multiplier > 128 * 128 * 128)
{
throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
}
multiplier *= 128;
} while ((encodedByte & 128) != 0);
return value;
@@ -42,13 +42,12 @@ namespace MQTTnet.Serializer
}
}
public MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body)
public MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body)
{
if (header == null) throw new ArgumentNullException(nameof(header));
if (body == null) throw new ArgumentNullException(nameof(body));
using (var bodyStream = new MemoryStream(body))
using (var reader = new MqttPacketReader(header, bodyStream))
using (var reader = new MqttPacketReader(header, body))
{
return Deserialize(header, reader);
}
@@ -76,7 +75,7 @@ namespace MQTTnet.Serializer
}
}
private static MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
private MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
{
switch (header.ControlPacketType)
{
@@ -279,20 +278,22 @@ namespace MQTTnet.Serializer
return packet;
}
private static MqttBasePacket DeserializeConnAck(MqttPacketReader reader)
private MqttBasePacket DeserializeConnAck(MqttPacketReader reader)
{
var variableHeader1 = reader.ReadByte();
var variableHeader2 = reader.ReadByte();
var packet = new MqttConnAckPacket();
var packet = new MqttConnAckPacket
var firstByteReader = new ByteReader(reader.ReadByte());
if (ProtocolVersion == MqttProtocolVersion.V311)
{
IsSessionPresent = new ByteReader(variableHeader1).Read(),
ConnectReturnCode = (MqttConnectReturnCode)variableHeader2
};
packet.IsSessionPresent = firstByteReader.Read();
}
packet.ConnectReturnCode = (MqttConnectReturnCode)reader.ReadByte();
return packet;
}
private static void ValidateConnectPacket(MqttConnectPacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
@@ -374,14 +375,21 @@ namespace MQTTnet.Serializer
private byte Serialize(MqttConnAckPacket packet, MqttPacketWriter writer)
{
var connectAcknowledgeFlags = new ByteWriter();
if (ProtocolVersion == MqttProtocolVersion.V311)
if (ProtocolVersion == MqttProtocolVersion.V310)
{
writer.Write(0);
}
else if (ProtocolVersion == MqttProtocolVersion.V311)
{
var connectAcknowledgeFlags = new ByteWriter();
connectAcknowledgeFlags.Write(packet.IsSessionPresent);
writer.Write(connectAcknowledgeFlags);
}
else
{
throw new MqttProtocolViolationException("Protocol version not supported.");
}
writer.Write(connectAcknowledgeFlags);
writer.Write((byte)packet.ConnectReturnCode);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
@@ -389,7 +397,12 @@ namespace MQTTnet.Serializer
private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter writer)
{
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("PubRel packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
}
@@ -402,7 +415,12 @@ namespace MQTTnet.Serializer
if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("Publish packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
}
else
{
@@ -436,21 +454,36 @@ namespace MQTTnet.Serializer
private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer)
{
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("PubAck packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
}
private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter writer)
{
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("PubRec packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
}
private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter writer)
{
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("PubComp packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
}
@@ -459,7 +492,12 @@ namespace MQTTnet.Serializer
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("Subscribe packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
if (packet.TopicFilters?.Count > 0)
{
@@ -475,7 +513,12 @@ namespace MQTTnet.Serializer
private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter writer)
{
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("SubAck packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
if (packet.SubscribeReturnCodes?.Any() == true)
{
@@ -492,7 +535,12 @@ namespace MQTTnet.Serializer
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("Unsubscribe packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
if (packet.TopicFilters?.Any() == true)
{
@@ -505,9 +553,14 @@ namespace MQTTnet.Serializer
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
}
private static byte Serialize(IMqttPacketWithIdentifier packet, BinaryWriter writer)
private static byte Serialize(MqttUnsubAckPacket packet, BinaryWriter writer)
{
writer.Write(packet.PacketIdentifier);
if (!packet.PacketIdentifier.HasValue)
{
throw new MqttProtocolViolationException("UnsubAck packet has no packet identifier.");
}
writer.Write(packet.PacketIdentifier.Value);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck);
}
@@ -12,5 +12,7 @@ namespace MQTTnet.Server
public TimeSpan LastPacketReceived { get; set; }
public TimeSpan LastNonKeepAlivePacketReceived { get; set; }
public int PendingApplicationMessages { get; set; }
}
}
@@ -6,12 +6,20 @@ namespace MQTTnet.Server
{
public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher
{
event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
event EventHandler<MqttServerStartedEventArgs> Started;
event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
IMqttServerOptions Options { get; }
Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();
Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters);
Task UnsubscribeAsync(string clientId, IList<string> topicFilters);
Task StartAsync(IMqttServerOptions options);
Task StopAsync();
}
@@ -1,5 +1,4 @@
using System;
using System.Net;
namespace MQTTnet.Server
{
@@ -0,0 +1,98 @@
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class MqttClientKeepAliveMonitor
{
private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch();
private readonly string _clientId;
private readonly Func<Task> _timeoutCallback;
private readonly IMqttNetLogger _logger;
private Task _workerTask;
public MqttClientKeepAliveMonitor(string clientId, Func<Task> timeoutCallback, IMqttNetLogger logger)
{
_clientId = clientId;
_timeoutCallback = timeoutCallback;
_logger = logger;
}
public TimeSpan LastPacketReceived => _lastPacketReceivedTracker.Elapsed;
public TimeSpan LastNonKeepAlivePacketReceived => _lastNonKeepAlivePacketReceivedTracker.Elapsed;
public void Start(int keepAlivePeriod, CancellationToken cancellationToken)
{
if (keepAlivePeriod == 0)
{
return;
}
_workerTask = Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken).ConfigureAwait(false), cancellationToken);
}
public void WaitForCompletion()
{
if (_workerTask != null)
{
Task.WaitAll(_workerTask);
}
}
private async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken)
{
try
{
_lastPacketReceivedTracker.Restart();
_lastNonKeepAlivePacketReceivedTracker.Restart();
while (!cancellationToken.IsCancellationRequested)
{
// Values described here: [MQTT-3.1.2-24].
if (_lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod * 1.5D)
{
_logger.Warning<MqttClientSession>("Client '{0}': Did not receive any packet or keep alive signal.", _clientId);
if (_timeoutCallback != null)
{
await _timeoutCallback().ConfigureAwait(false);
}
return;
}
await Task.Delay(keepAlivePeriod, cancellationToken).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", _clientId);
}
finally
{
_logger.Trace<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", _clientId);
}
}
public void PacketReceived(MqttBasePacket packet)
{
_lastPacketReceivedTracker.Restart();
if (!(packet is MqttPingReqPacket))
{
_lastNonKeepAlivePacketReceivedTracker.Restart();
}
}
}
}
@@ -1,5 +1,6 @@
using System;
using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
@@ -12,18 +13,23 @@ namespace MQTTnet.Server
{
public sealed class MqttClientPendingMessagesQueue : IDisposable
{
private readonly BlockingCollection<MqttBasePacket> _queue = new BlockingCollection<MqttBasePacket>();
private readonly ConcurrentQueue<MqttBasePacket> _queue = new ConcurrentQueue<MqttBasePacket>();
private readonly SemaphoreSlim _queueWaitSemaphore = new SemaphoreSlim(0);
private readonly IMqttServerOptions _options;
private readonly MqttClientSession _session;
private readonly MqttClientSession _clientSession;
private readonly IMqttNetLogger _logger;
public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession session, IMqttNetLogger logger)
private Task _workerTask;
public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_session = session ?? throw new ArgumentNullException(nameof(session));
_options = options ?? throw new ArgumentNullException(nameof(options));
_clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
public int Count => _queue.Count;
public void Start(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -33,15 +39,25 @@ namespace MQTTnet.Server
return;
}
Task.Run(async () => await SendQueuedPacketsAsync(adapter, cancellationToken), cancellationToken).ConfigureAwait(false);
_workerTask = Task.Run(() => SendQueuedPacketsAsync(adapter, cancellationToken), cancellationToken);
}
public void WaitForCompletion()
{
if (_workerTask != null)
{
Task.WaitAll(_workerTask);
}
}
public void Enqueue(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
_queue.Add(packet);
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _session.ClientId);
_queue.Enqueue(packet);
_queueWaitSemaphore.Release();
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
}
private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -50,7 +66,7 @@ namespace MQTTnet.Server
{
while (!cancellationToken.IsCancellationRequested)
{
await SendQueuedPacketAsync(adapter, cancellationToken);
await SendNextQueuedPacketAsync(adapter, cancellationToken).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
@@ -58,36 +74,46 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Error<MqttClientPendingMessagesQueue>(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _session.ClientId);
_logger.Error<MqttClientPendingMessagesQueue>(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _clientSession.ClientId);
}
}
private async Task SendQueuedPacketAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
private async Task SendNextQueuedPacketAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
MqttBasePacket packet = null;
try
{
packet = _queue.Take(cancellationToken);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
await _queueWaitSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
if (!_queue.TryDequeue(out packet))
{
throw new InvalidOperationException(); // should not happen
}
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _session.ClientId);
if (cancellationToken.IsCancellationRequested)
{
return;
}
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false);
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
}
catch (Exception exception)
{
if (exception is MqttCommunicationTimedOutException)
{
_logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _session.ClientId);
_logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _clientSession.ClientId);
}
else if (exception is MqttCommunicationException)
{
_logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _session.ClientId);
_logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _clientSession.ClientId);
}
else if (exception is OperationCanceledException)
{
}
else
{
_logger.Error<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed (ClientId: {0}).", _session.ClientId);
_logger.Error<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed (ClientId: {0}).", _clientSession.ClientId);
}
if (packet is MqttPublishPacket publishPacket)
@@ -95,17 +121,21 @@ namespace MQTTnet.Server
if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
publishPacket.Dup = true;
_queue.Add(packet, CancellationToken.None);
_queue.Enqueue(packet);
_queueWaitSemaphore.Release();
}
}
await _session.StopAsync();
if (!cancellationToken.IsCancellationRequested)
{
await _clientSession.StopAsync().ConfigureAwait(false);
}
}
}
public void Dispose()
{
_queue?.Dispose();
_queueWaitSemaphore?.Dispose();
}
}
}
@@ -1,8 +1,9 @@
using System;
using System.Diagnostics;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
@@ -14,16 +15,10 @@ namespace MQTTnet.Server
{
public sealed class MqttClientSession : IDisposable
{
private readonly Stopwatch _lastPacketReceivedTracker = Stopwatch.StartNew();
private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = Stopwatch.StartNew();
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
private readonly IMqttServerOptions _options;
private readonly IMqttNetLogger _logger;
private readonly MqttClientSessionsManager _sessionsManager;
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly MqttClientSubscriptionsManager _subscriptionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
private IMqttChannelAdapter _adapter;
private CancellationTokenSource _cancellationTokenSource;
@@ -33,28 +28,31 @@ namespace MQTTnet.Server
string clientId,
IMqttServerOptions options,
MqttRetainedMessagesManager retainedMessagesManager,
MqttClientSessionsManager sessionsManager,
IMqttNetLogger logger)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
_sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
ClientId = clientId;
_subscriptionsManager = new MqttClientSubscriptionsManager(_options);
_pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeoutAsync, _logger);
SubscriptionsManager = new MqttClientSubscriptionsManager(_options, clientId);
PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
}
public Func<MqttClientSession, MqttApplicationMessage, Task> ApplicationMessageReceivedCallback { get; set; }
public MqttClientSubscriptionsManager SubscriptionsManager { get; }
public MqttClientPendingMessagesQueue PendingMessagesQueue { get; }
public MqttClientKeepAliveMonitor KeepAliveMonitor { get; }
public string ClientId { get; }
public MqttProtocolVersion? ProtocolVersion => _adapter?.PacketSerializer.ProtocolVersion;
public TimeSpan LastPacketReceived => _lastPacketReceivedTracker.Elapsed;
public TimeSpan LastNonKeepAlivePacketReceived => _lastNonKeepAlivePacketReceivedTracker.Elapsed;
public bool IsConnected => _adapter != null;
public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
@@ -70,15 +68,8 @@ namespace MQTTnet.Server
_adapter = adapter;
_cancellationTokenSource = cancellationTokenSource;
_pendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
_lastPacketReceivedTracker.Restart();
_lastNonKeepAlivePacketReceivedTracker.Restart();
if (connectPacket.KeepAlivePeriod > 0)
{
StartCheckingKeepAliveTimeout(TimeSpan.FromSeconds(connectPacket.KeepAlivePeriod), cancellationTokenSource.Token);
}
PendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
KeepAliveMonitor.Start(connectPacket.KeepAlivePeriod, cancellationTokenSource.Token);
await ReceivePacketsAsync(adapter, cancellationTokenSource.Token).ConfigureAwait(false);
}
@@ -105,14 +96,14 @@ namespace MQTTnet.Server
}
_cancellationTokenSource?.Cancel(false);
PendingMessagesQueue.WaitForCompletion();
KeepAliveMonitor.WaitForCompletion();
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
if (_adapter != null)
{
await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
_adapter = null;
}
_adapter = null;
_logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId);
}
@@ -121,8 +112,8 @@ namespace MQTTnet.Server
var willMessage = _willMessage;
if (willMessage != null)
{
_willMessage = null; //clear willmessage so it is send just once
await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage).ConfigureAwait(false);
_willMessage = null; // clear willmessage so it is send just once
await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
}
}
}
@@ -131,7 +122,7 @@ namespace MQTTnet.Server
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
var result = await _subscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
var result = await SubscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
if (!result.IsSubscribed)
{
return;
@@ -140,15 +131,52 @@ namespace MQTTnet.Server
var publishPacket = applicationMessage.ToPublishPacket();
publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
_pendingMessagesQueue.Enqueue(publishPacket);
if (publishPacket.QualityOfServiceLevel > 0)
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
}
PendingMessagesQueue.Enqueue(publishPacket);
}
public async Task SubscribeAsync(IList<TopicFilter> topicFilters)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
await SubscriptionsManager.SubscribeAsync(new MqttSubscribePacket
{
TopicFilters = topicFilters
}).ConfigureAwait(false);
await EnqueueSubscribedRetainedMessagesAsync(topicFilters).ConfigureAwait(false);
}
public Task UnsubscribeAsync(IList<string> topicFilters)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
return SubscriptionsManager.UnsubscribeAsync(new MqttUnsubscribePacket
{
TopicFilters = topicFilters
});
}
public void Dispose()
{
_pendingMessagesQueue?.Dispose();
ApplicationMessageReceivedCallback = null;
SubscriptionsManager?.Dispose();
PendingMessagesQueue?.Dispose();
_cancellationTokenSource?.Dispose();
}
private Task StopDueToKeepAliveTimeoutAsync()
{
_logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
return StopAsync();
}
private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
try
@@ -156,14 +184,7 @@ namespace MQTTnet.Server
while (!cancellationToken.IsCancellationRequested)
{
var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
_lastPacketReceivedTracker.Restart();
if (!(packet is MqttPingReqPacket))
{
_lastNonKeepAlivePacketReceivedTracker.Restart();
}
KeepAliveMonitor.PacketReceived(packet);
await ProcessReceivedPacketAsync(adapter, packet, cancellationToken).ConfigureAwait(false);
}
}
@@ -191,7 +212,7 @@ namespace MQTTnet.Server
if (packet is MqttPingReqPacket)
{
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPingRespPacket());
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttPingRespPacket() });
}
if (packet is MqttPubRelPacket pubRelPacket)
@@ -201,7 +222,12 @@ namespace MQTTnet.Server
if (packet is MqttPubRecPacket pubRecPacket)
{
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, pubRecPacket.CreateResponse<MqttPubRelPacket>());
var responsePacket = new MqttPubRelPacket
{
PacketIdentifier = pubRecPacket.PacketIdentifier
};
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { responsePacket });
}
if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
@@ -231,31 +257,30 @@ namespace MQTTnet.Server
private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = await _subscriptionsManager.SubscribeAsync(subscribePacket, ClientId);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false);
var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { subscribeResult.ResponsePacket }).ConfigureAwait(false);
if (subscribeResult.CloseConnection)
{
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttDisconnectPacket() }).ConfigureAwait(false);
await StopAsync().ConfigureAwait(false);
}
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
}
private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
var unsubscribeResult = await _subscriptionsManager.UnsubscribeAsync(unsubscribePacket);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult);
var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { unsubscribeResult });
}
private async Task EnqueueSubscribedRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
{
var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(subscribePacket);
foreach (var publishPacket in retainedMessages)
var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters);
foreach (var applicationMessage in retainedMessages)
{
await EnqueueApplicationMessageAsync(publishPacket);
await EnqueueApplicationMessageAsync(applicationMessage).ConfigureAwait(false);
}
}
@@ -267,7 +292,7 @@ namespace MQTTnet.Server
{
case MqttQualityOfServiceLevel.AtMostOnce:
{
return _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
return ApplicationMessageReceivedCallback?.Invoke(this, applicationMessage);
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
@@ -286,62 +311,25 @@ namespace MQTTnet.Server
private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
}
private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
// QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
}
private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{
var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier };
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response);
}
private void StartCheckingKeepAliveTimeout(TimeSpan keepAlivePeriod, CancellationToken cancellationToken)
{
Task.Run(
async () => await CheckKeepAliveTimeoutAsync(keepAlivePeriod, cancellationToken).ConfigureAwait(false)
, cancellationToken);
}
private async Task CheckKeepAliveTimeoutAsync(TimeSpan keepAlivePeriod, CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
// Values described here: [MQTT-3.1.2-24].
if (_lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod.TotalSeconds * 1.5D)
{
_logger.Warning<MqttClientSession>("Client '{0}': Did not receive any packet or keep alive signal.", ClientId);
await StopAsync();
return;
}
await Task.Delay(keepAlivePeriod, cancellationToken);
}
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", ClientId);
}
finally
{
_logger.Trace<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", ClientId);
}
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response });
}
}
}
@@ -12,32 +12,39 @@ using MQTTnet.Serializer;
namespace MQTTnet.Server
{
public sealed class MqttClientSessionsManager
public sealed class MqttClientSessionsManager : IDisposable
{
private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly IMqttServerOptions _options;
private readonly MqttServer _server;
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly IMqttNetLogger _logger;
public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, MqttServer server, IMqttNetLogger logger)
public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
{
_server = server ?? throw new ArgumentNullException(nameof(server));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
}
public async Task RunClientSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
public Action<ConnectedMqttClient> ClientConnectedCallback { get; set; }
public Action<ConnectedMqttClient> ClientDisconnectedCallback { get; set; }
public Action<string, TopicFilter> ClientSubscribedTopicCallback { get; set; }
public Action<string, string> ClientUnsubscribedTopicCallback { get; set; }
public Action<string, MqttApplicationMessage> ApplicationMessageReceivedCallback { get; set; }
public async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
{
var clientId = string.Empty;
MqttClientSession clientSession = null;
try
{
if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false) is MqttConnectPacket connectPacket))
if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken)
.ConfigureAwait(false) is MqttConnectPacket connectPacket))
{
throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
throw new MqttProtocolViolationException(
"The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
}
clientId = connectPacket.ClientId;
@@ -48,29 +55,39 @@ namespace MQTTnet.Server
var connectReturnCode = ValidateConnection(connectPacket);
if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
{
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
{
ConnectReturnCode = connectReturnCode
new MqttConnAckPacket
{
ConnectReturnCode = connectReturnCode
}
}).ConfigureAwait(false);
return;
}
var clientSession = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
clientSession = result.Session;
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
{
ConnectReturnCode = connectReturnCode,
IsSessionPresent = clientSession.IsExistingSession
new MqttConnAckPacket
{
ConnectReturnCode = connectReturnCode,
IsSessionPresent = result.IsExistingSession
}
}).ConfigureAwait(false);
_server.OnClientConnected(new ConnectedMqttClient
ClientConnectedCallback?.Invoke(new ConnectedMqttClient
{
ClientId = clientId,
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
});
await clientSession.Session.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
@@ -81,16 +98,18 @@ namespace MQTTnet.Server
try
{
await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
clientAdapter.Dispose();
}
catch (Exception)
catch (Exception exception)
{
// ignored
_logger.Error<MqttClientSessionsManager>(exception, exception.Message);
}
_server.OnClientDisconnected(new ConnectedMqttClient
ClientDisconnectedCallback?.Invoke(new ConnectedMqttClient
{
ClientId = clientId,
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion,
PendingApplicationMessages = clientSession?.PendingMessagesQueue.Count ?? 0
});
}
}
@@ -122,8 +141,9 @@ namespace MQTTnet.Server
{
ClientId = s.Value.ClientId,
ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311,
LastPacketReceived = s.Value.LastPacketReceived,
LastNonKeepAlivePacketReceived = s.Value.LastNonKeepAlivePacketReceived
LastPacketReceived = s.Value.KeepAliveMonitor.LastPacketReceived,
LastNonKeepAlivePacketReceived = s.Value.KeepAliveMonitor.LastNonKeepAlivePacketReceived,
PendingApplicationMessages = s.Value.PendingMessagesQueue.Count
}).ToList();
}
finally
@@ -147,7 +167,7 @@ namespace MQTTnet.Server
await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
}
_server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage);
ApplicationMessageReceivedCallback?.Invoke(senderClientSession?.ClientId, applicationMessage);
}
catch (Exception exception)
{
@@ -168,18 +188,61 @@ namespace MQTTnet.Server
}
}
public async Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
{
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
if (!_sessions.TryGetValue(clientId, out var session))
{
throw new InvalidOperationException($"Client session {clientId} is unknown.");
}
await session.SubscribeAsync(topicFilters);
}
finally
{
_semaphore.Release();
}
}
public async Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
{
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
if (!_sessions.TryGetValue(clientId, out var session))
{
throw new InvalidOperationException($"Client session {clientId} is unknown.");
}
await session.UnsubscribeAsync(topicFilters);
}
finally
{
_semaphore.Release();
}
}
private MqttApplicationMessage InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
{
if (_options.ApplicationMessageInterceptor == null)
var interceptor = _options.ApplicationMessageInterceptor;
if (interceptor == null)
{
return applicationMessage;
}
var interceptorContext = new MqttApplicationMessageInterceptorContext(
senderClientSession.ClientId,
senderClientSession?.ClientId,
applicationMessage);
_options.ApplicationMessageInterceptor(interceptorContext);
interceptor(interceptorContext);
return interceptorContext.ApplicationMessage;
}
@@ -229,7 +292,14 @@ namespace MQTTnet.Server
{
isExistingSession = false;
clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, this, _logger);
clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, _logger)
{
ApplicationMessageReceivedCallback = DispatchApplicationMessageAsync
};
clientSession.SubscriptionsManager.TopicSubscribedCallback = ClientSubscribedTopicCallback;
clientSession.SubscriptionsManager.TopicUnsubscribedCallback = ClientUnsubscribedTopicCallback;
_sessions[connectPacket.ClientId] = clientSession;
_logger.Trace<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
@@ -242,5 +312,16 @@ namespace MQTTnet.Server
_semaphore.Release();
}
}
public void Dispose()
{
ClientConnectedCallback = null;
ClientDisconnectedCallback = null;
ClientSubscribedTopicCallback = null;
ClientUnsubscribedTopicCallback = null;
ApplicationMessageReceivedCallback = null;
_semaphore?.Dispose();
}
}
}
@@ -0,0 +1,17 @@
using System;
namespace MQTTnet.Server
{
public class MqttClientSubscribedTopicEventArgs : EventArgs
{
public MqttClientSubscribedTopicEventArgs(string clientId, TopicFilter topicFilter)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
}
public string ClientId { get; }
public TopicFilter TopicFilter { get; }
}
}
@@ -8,24 +8,33 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttClientSubscriptionsManager
public sealed class MqttClientSubscriptionsManager : IDisposable
{
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly IMqttServerOptions _options;
private readonly string _clientId;
public MqttClientSubscriptionsManager(IMqttServerOptions options)
public MqttClientSubscriptionsManager(IMqttServerOptions options, string clientId)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
}
public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket, string clientId)
public Action<string, TopicFilter> TopicSubscribedCallback { get; set; }
public Action<string, string> TopicUnsubscribedCallback { get; set; }
public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket)
{
if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
var result = new MqttClientSubscribeResult
{
ResponsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>(),
ResponsePacket = new MqttSubAckPacket
{
PacketIdentifier = subscribePacket.PacketIdentifier
},
CloseConnection = false
};
@@ -34,7 +43,7 @@ namespace MQTTnet.Server
{
foreach (var topicFilter in subscribePacket.TopicFilters)
{
var interceptorContext = InterceptSubscribe(clientId, topicFilter);
var interceptorContext = InterceptSubscribe(topicFilter);
if (!interceptorContext.AcceptSubscription)
{
result.ResponsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.Failure);
@@ -52,6 +61,7 @@ namespace MQTTnet.Server
if (interceptorContext.AcceptSubscription)
{
_subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
TopicSubscribedCallback?.Invoke(_clientId, topicFilter);
}
}
}
@@ -73,6 +83,7 @@ namespace MQTTnet.Server
foreach (var topicFilter in unsubscribePacket.TopicFilters)
{
_subscriptions.Remove(topicFilter);
TopicUnsubscribedCallback?.Invoke(_clientId, topicFilter);
}
}
finally
@@ -80,7 +91,10 @@ namespace MQTTnet.Server
_semaphore.Release();
}
return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>();
return new MqttUnsubAckPacket
{
PacketIdentifier = unsubscribePacket.PacketIdentifier
};
}
public async Task<CheckSubscriptionsResult> CheckSubscriptionsAsync(MqttApplicationMessage applicationMessage)
@@ -117,9 +131,25 @@ namespace MQTTnet.Server
}
}
private MqttSubscriptionInterceptorContext InterceptSubscribe(string clientId, TopicFilter topicFilter)
public void Dispose()
{
var interceptorContext = new MqttSubscriptionInterceptorContext(clientId, topicFilter);
_semaphore?.Dispose();
}
private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
{
switch (qualityOfServiceLevel)
{
case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
default: return MqttSubscribeReturnCode.Failure;
}
}
private MqttSubscriptionInterceptorContext InterceptSubscribe(TopicFilter topicFilter)
{
var interceptorContext = new MqttSubscriptionInterceptorContext(_clientId, topicFilter);
_options.SubscriptionInterceptor?.Invoke(interceptorContext);
return interceptorContext;
}
@@ -146,16 +176,5 @@ namespace MQTTnet.Server
QualityOfServiceLevel = effectiveQoS
};
}
private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
{
switch (qualityOfServiceLevel)
{
case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
default: return MqttSubscribeReturnCode.Failure;
}
}
}
}
@@ -0,0 +1,17 @@
using System;
namespace MQTTnet.Server
{
public class MqttClientUnsubscribedTopicEventArgs : EventArgs
{
public MqttClientUnsubscribedTopicEventArgs(string clientId, string topicFilter)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
}
public string ClientId { get; }
public string TopicFilter { get; }
}
}
@@ -4,11 +4,10 @@ using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class MqttRetainedMessagesManager
public sealed class MqttRetainedMessagesManager : IDisposable
{
private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
@@ -68,7 +67,7 @@ namespace MQTTnet.Server
}
}
public async Task<List<MqttApplicationMessage>> GetSubscribedMessagesAsync(MqttSubscribePacket subscribePacket)
public async Task<List<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters)
{
var retainedMessages = new List<MqttApplicationMessage>();
@@ -77,13 +76,8 @@ namespace MQTTnet.Server
{
foreach (var retainedMessage in _retainedMessages.Values)
{
foreach (var topicFilter in subscribePacket.TopicFilters)
foreach (var topicFilter in topicFilters)
{
if (retainedMessage.QualityOfServiceLevel < topicFilter.QualityOfServiceLevel)
{
continue;
}
if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic))
{
continue;
@@ -102,6 +96,11 @@ namespace MQTTnet.Server
return retainedMessages;
}
public void Dispose()
{
_semaphore?.Dispose();
}
private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
{
var saveIsRequired = false;
@@ -16,30 +16,47 @@ namespace MQTTnet.Server
private MqttClientSessionsManager _clientSessionsManager;
private MqttRetainedMessagesManager _retainedMessagesManager;
private CancellationTokenSource _cancellationTokenSource;
private IMqttServerOptions _options;
public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
{
if (adapters == null) throw new ArgumentNullException(nameof(adapters));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
if (adapters == null)
{
throw new ArgumentNullException(nameof(adapters));
}
_adapters = adapters.ToList();
}
public event EventHandler<MqttServerStartedEventArgs> Started;
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public IMqttServerOptions Options { get; private set; }
public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
{
return _clientSessionsManager.GetConnectedClientsAsync();
}
public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
{
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
return _clientSessionsManager.SubscribeAsync(clientId, topicFilters);
}
public Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
{
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
}
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
@@ -54,20 +71,28 @@ namespace MQTTnet.Server
public async Task StartAsync(IMqttServerOptions options)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
Options = options ?? throw new ArgumentNullException(nameof(options));
if (_cancellationTokenSource != null) throw new InvalidOperationException("The server is already started.");
_cancellationTokenSource = new CancellationTokenSource();
_retainedMessagesManager = new MqttRetainedMessagesManager(_options, _logger);
_clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, this, _logger);
_retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
await _retainedMessagesManager.LoadMessagesAsync();
_clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _logger)
{
ClientConnectedCallback = OnClientConnected,
ClientDisconnectedCallback = OnClientDisconnected,
ClientSubscribedTopicCallback = OnClientSubscribedTopic,
ClientUnsubscribedTopicCallback = OnClientUnsubscribedTopic,
ApplicationMessageReceivedCallback = OnApplicationMessageReceived
};
foreach (var adapter in _adapters)
{
adapter.ClientAccepted += OnClientAccepted;
await adapter.StartAsync(_options);
await adapter.StartAsync(Options);
}
_logger.Info<MqttServer>("Started.");
@@ -98,39 +123,46 @@ namespace MQTTnet.Server
}
finally
{
_clientSessionsManager?.Dispose();
_retainedMessagesManager?.Dispose();
_cancellationTokenSource = null;
_retainedMessagesManager = null;
_clientSessionsManager = null;
}
}
internal void OnClientConnected(ConnectedMqttClient client)
private void OnClientConnected(ConnectedMqttClient client)
{
if (client == null) throw new ArgumentNullException(nameof(client));
_logger.Info<MqttServer>("Client '{0}': Connected.", client.ClientId);
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
}
internal void OnClientDisconnected(ConnectedMqttClient client)
private void OnClientDisconnected(ConnectedMqttClient client)
{
if (client == null) throw new ArgumentNullException(nameof(client));
_logger.Info<MqttServer>("Client '{0}': Disconnected.", client.ClientId);
ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client));
}
internal void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
private void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
}
private void OnClientUnsubscribedTopic(string clientId, string topicFilter)
{
ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
}
private void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
{
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));
}
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
{
eventArgs.SessionTask = Task.Run(
async () => await _clientSessionsManager.RunClientSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
async () => await _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
_cancellationTokenSource.Token);
}
}
@@ -0,0 +1,65 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">netstandard1.3;netstandard2.0;net452;net461;uap10.0</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' != 'Windows_NT' ">netstandard1.3;netstandard2.0</TargetFrameworks>
<AssemblyName>MQTTnet</AssemblyName>
<RootNamespace>MQTTnet</RootNamespace>
<GeneratePackageOnBuild>False</GeneratePackageOnBuild>
<DebugType>Full</DebugType>
<AssemblyVersion>0.0.0.0</AssemblyVersion>
<FileVersion>0.0.0.0</FileVersion>
<Version>0.0.0.0</Version>
<Company />
<Product />
<Description />
<Authors />
<PackageId />
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
</PropertyGroup>
<PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
<CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
<NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion>10.0.16299.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
<DefaultLanguage>en</DefaultLanguage>
<LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x86'" />
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" />
<PropertyGroup Condition="'$(Configuration)|$(TargetFramework)|$(Platform)'=='Release|netstandard1.3|AnyCPU'">
<DefineConstants>RELEASE;NETSTANDARD1_3</DefineConstants>
</PropertyGroup>
<ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
<PackageReference Include="System.Net.Security" Version="4.3.2" />
<PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
<PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">
<PackageReference Include="System.Net.Security" Version="4.3.2" />
<PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
<PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="5.4.1" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net452'">
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net461'">
</ItemGroup>
</Project>
+6
View File
@@ -1,7 +1,13 @@
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSORHOLDER_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_CALL_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpRenamePlacementToArrangementMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EAddAccessorOwnerDeclarationBracesMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+3 -2
View File
@@ -18,12 +18,12 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
* TLS 1.2 support for client and server (but not UWP servers)
* Extensible communication channels (i.e. In-Memory, TCP, TCP+TLS, WS)
* Lightweight (only the low level implementation of MQTT, no overhead)
* Performance optimized (processing ~30.000 messages / second)*
* Performance optimized (processing ~40.000 messages / second)*
* Interfaces included for mocking and testing
* Access to internal trace messages
* Unit tested (70+ tests)
\* Tested on local machine 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_.
\* 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_.
### Client
@@ -43,6 +43,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
* Retained messages are supported including persisting via interface methods (own implementation required)
* WebSockets supported (via ASP.NET Core 2.0, separate nuget)
* A custom message interceptor can be added which allows transforming or extending every received application message
* Validate subscriptions and deny subscribing of certain topics depending on requesting clients
## Supported frameworks
@@ -0,0 +1,36 @@
using System.Net.Sockets;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Exceptions;
namespace MQTTnet.Core.Tests
{
[TestClass]
public class MqttClientTests
{
[TestMethod]
public async Task ClientDisconnectException()
{
var factory = new MqttFactory();
var client = factory.CreateMqttClient();
var exceptionIsCorrect = false;
client.Disconnected += (s, e) =>
{
exceptionIsCorrect = e.Exception is MqttCommunicationException c && c.InnerException is SocketException;
};
try
{
await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("wrong-server").Build());
}
catch
{
}
Assert.IsTrue(exceptionIsCorrect);
}
}
}
@@ -0,0 +1,66 @@
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Diagnostics;
using MQTTnet.Packets;
using MQTTnet.Server;
namespace MQTTnet.Core.Tests
{
[TestClass]
public class MqttKeepAliveMonitorTests
{
[TestMethod]
public void KeepAlive_Timeout()
{
var timeoutCalledCount = 0;
var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate
{
timeoutCalledCount++;
return Task.FromResult(0);
}, new MqttNetLogger());
Assert.AreEqual(0, timeoutCalledCount);
monitor.Start(1, CancellationToken.None);
Assert.AreEqual(0, timeoutCalledCount);
Thread.Sleep(2000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.
Assert.AreEqual(1, timeoutCalledCount);
}
[TestMethod]
public void KeepAlive_NoTimeout()
{
var timeoutCalledCount = 0;
var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate
{
timeoutCalledCount++;
return Task.FromResult(0);
}, new MqttNetLogger());
Assert.AreEqual(0, timeoutCalledCount);
monitor.Start(1, CancellationToken.None);
Assert.AreEqual(0, timeoutCalledCount);
// Simulate traffic.
Thread.Sleep(1000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.
monitor.PacketReceived(new MqttPublishPacket());
Thread.Sleep(1000);
monitor.PacketReceived(new MqttPublishPacket());
Thread.Sleep(1000);
Assert.AreEqual(0, timeoutCalledCount);
Thread.Sleep(2000);
Assert.AreEqual(1, timeoutCalledCount);
}
}
}
@@ -0,0 +1,32 @@
using Microsoft.VisualStudio.TestTools .UnitTesting;
using MQTTnet.Client;
namespace MQTTnet.Core.Tests
{
[TestClass]
public class MqttPacketIdentifierProviderTests
{
[TestMethod]
public void Reset()
{
var p = new MqttPacketIdentifierProvider();
Assert.AreEqual(1, p.GetNewPacketIdentifier());
Assert.AreEqual(2, p.GetNewPacketIdentifier());
p.Reset();
Assert.AreEqual(1, p.GetNewPacketIdentifier());
}
[TestMethod]
public void ReachBoundaries()
{
var p = new MqttPacketIdentifierProvider();
for (ushort i = 0; i < ushort.MaxValue; i++)
{
Assert.AreEqual(i + 1, p.GetNewPacketIdentifier());
}
Assert.AreEqual(1, p.GetNewPacketIdentifier());
}
}
}
@@ -12,7 +12,7 @@ namespace MQTTnet.Core.Tests
public void MqttPacketReader_EmptyStream()
{
var memStream = new MemoryStream();
var header = MqttPacketReader.ReadHeaderFromSource(memStream, CancellationToken.None);
var header = MqttPacketReader.ReadHeaderAsync(memStream, CancellationToken.None).GetAwaiter().GetResult();
Assert.IsNull(header);
}
@@ -119,10 +119,10 @@ namespace MQTTnet.Core.Tests
{
var p = new MqttConnAckPacket
{
ConnectReturnCode = MqttConnectReturnCode.ConnectionAccepted
ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized
};
SerializeAndCompare(p, "IAIAAA==", MqttProtocolVersion.V310);
SerializeAndCompare(p, "IAIABQ==", MqttProtocolVersion.V310);
}
[TestMethod]
@@ -137,6 +137,17 @@ namespace MQTTnet.Core.Tests
DeserializeAndCompare(p, "IAIBBQ==");
}
[TestMethod]
public void DeserializeV310_MqttConnAckPacket()
{
var p = new MqttConnAckPacket
{
ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized
};
DeserializeAndCompare(p, "IAIABQ==", MqttProtocolVersion.V310);
}
[TestMethod]
public void SerializeV311_MqttDisconnectPacket()
{
@@ -397,19 +408,19 @@ namespace MQTTnet.Core.Tests
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(chunks)));
}
private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
{
var serializer = new MqttPacketSerializer();
var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion };
var buffer1 = serializer.Serialize(packet);
using (var headerStream = new MemoryStream(Join(buffer1)))
{
var header = MqttPacketReader.ReadHeaderFromSource(headerStream, CancellationToken.None);
var header = MqttPacketReader.ReadHeaderAsync(headerStream, CancellationToken.None).GetAwaiter().GetResult();
using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.BodyLength))
{
var deserializedPacket = serializer.Deserialize(header, bodyStream.ToArray());
var deserializedPacket = serializer.Deserialize(header, bodyStream);
var buffer2 = serializer.Serialize(deserializedPacket);
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));
+56 -3
View File
@@ -79,7 +79,7 @@ namespace MQTTnet.Core.Tests
}
[TestMethod]
public async Task MqttServer_Unsubscribe()
public async Task MqttServer_SubscribeUnsubscribe()
{
var serverAdapter = new TestMqttServerAdapter();
var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
@@ -99,13 +99,31 @@ namespace MQTTnet.Core.Tests
await c2.PublishAsync(message);
Assert.AreEqual(0, receivedMessagesCount);
var subscribeEventCalled = false;
s.ClientSubscribedTopic += (_, e) =>
{
subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == "c1";
};
await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
await Task.Delay(100);
Assert.IsTrue(subscribeEventCalled, "Subscribe event not called.");
await c2.PublishAsync(message);
await Task.Delay(500);
Assert.AreEqual(1, receivedMessagesCount);
var unsubscribeEventCalled = false;
s.ClientUnsubscribedTopic += (_, e) =>
{
unsubscribeEventCalled = e.TopicFilter == "a" && e.ClientId == "c1";
};
await c1.UnsubscribeAsync("a");
await Task.Delay(100);
Assert.IsTrue(unsubscribeEventCalled, "Unsubscribe event not called.");
await c2.PublishAsync(message);
await Task.Delay(500);
@@ -138,7 +156,7 @@ namespace MQTTnet.Core.Tests
var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
s.PublishAsync(message).Wait();
await s.PublishAsync(message);
await Task.Delay(500);
}
finally
@@ -299,7 +317,7 @@ namespace MQTTnet.Core.Tests
var serverAdapter = new TestMqttServerAdapter();
var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
try
{
var options = new MqttServerOptions { ApplicationMessageInterceptor = Interceptor };
@@ -330,6 +348,41 @@ namespace MQTTnet.Core.Tests
}
}
[TestMethod]
public async Task MqttServer_Body()
{
var serverAdapter = new TestMqttServerAdapter();
var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
var bodyIsMatching = false;
try
{
await s.StartAsync(new MqttServerOptions());
var c1 = await serverAdapter.ConnectTestClient(s, "c1");
var c2 = await serverAdapter.ConnectTestClient(s, "c2");
c1.ApplicationMessageReceived += (_, e) =>
{
if (Encoding.UTF8.GetString(e.ApplicationMessage.Payload) == "The body")
{
bodyIsMatching = true;
}
};
await c1.SubscribeAsync("A", MqttQualityOfServiceLevel.AtMostOnce);
await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("A").WithPayload(Encoding.UTF8.GetBytes("The body")).Build());
await Task.Delay(500);
}
finally
{
await s.StopAsync();
}
Assert.IsTrue(bodyIsMatching);
}
private class TestStorage : IMqttServerStorage
{
private IList<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();
@@ -11,12 +11,12 @@ namespace MQTTnet.Core.Tests
[TestMethod]
public void MqttSubscriptionsManager_SubscribeSingleSuccess()
{
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
sm.SubscribeAsync(sp, "").Wait();
sm.SubscribeAsync(sp).Wait();
var pp = new MqttApplicationMessage
{
@@ -32,12 +32,12 @@ namespace MQTTnet.Core.Tests
[TestMethod]
public void MqttSubscriptionsManager_SubscribeDifferentQoSSuccess()
{
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce));
sm.SubscribeAsync(sp, "").Wait();
sm.SubscribeAsync(sp).Wait();
var pp = new MqttApplicationMessage
{
@@ -53,13 +53,13 @@ namespace MQTTnet.Core.Tests
[TestMethod]
public void MqttSubscriptionsManager_SubscribeTwoTimesSuccess()
{
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce));
sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtLeastOnce));
sm.SubscribeAsync(sp, "").Wait();
sm.SubscribeAsync(sp).Wait();
var pp = new MqttApplicationMessage
{
@@ -75,12 +75,12 @@ namespace MQTTnet.Core.Tests
[TestMethod]
public void MqttSubscriptionsManager_SubscribeSingleNoSuccess()
{
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
sm.SubscribeAsync(sp, "").Wait();
sm.SubscribeAsync(sp).Wait();
var pp = new MqttApplicationMessage
{
@@ -94,12 +94,12 @@ namespace MQTTnet.Core.Tests
[TestMethod]
public void MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle()
{
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
sm.SubscribeAsync(sp, "").Wait();
sm.SubscribeAsync(sp).Wait();
var pp = new MqttApplicationMessage
{
@@ -17,6 +17,10 @@ namespace MQTTnet.Core.Tests
public IMqttPacketSerializer PacketSerializer { get; } = new MqttPacketSerializer();
public void Dispose()
{
}
public Task ConnectAsync(TimeSpan timeout)
{
return Task.FromResult(0);
@@ -43,7 +47,17 @@ namespace MQTTnet.Core.Tests
{
ThrowIfPartnerIsNull();
return Task.Run(() => _incomingPackets.Take(), cancellationToken);
return Task.Run(() =>
{
try
{
return _incomingPackets.Take(cancellationToken);
}
catch
{
return null;
}
}, cancellationToken);
}
private void EnqueuePacketInternal(MqttBasePacket packet)
@@ -10,7 +10,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.3" />
<PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.5" />
</ItemGroup>
<ItemGroup>
@@ -6,7 +6,7 @@
},
"dependencies": {
"mqtt": "2.13.1",
"mqtt": "2.15.1",
"@types/node": "8.0.46",
"systemjs": "0.20.19",
"typescript": "2.3.4"
+1 -1
View File
@@ -89,7 +89,7 @@ namespace MQTTnet.TestApp.NetCore
c.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(content.ToString());
}
}
catch (Exception e)
catch (Exception)
{
}
};
@@ -12,7 +12,7 @@
<DefaultLanguage>en-US</DefaultLanguage>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion Condition=" '$(TargetPlatformVersion)' == '' ">10.0.16299.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.15063.0</TargetPlatformMinVersion>
<TargetPlatformMinVersion>10.0.16299.0</TargetPlatformMinVersion>
<MinimumVisualStudioVersion>14</MinimumVisualStudioVersion>
<FileAlignment>512</FileAlignment>
<ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
@@ -138,10 +138,10 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
<Version>5.4.0</Version>
<Version>6.0.8</Version>
</PackageReference>
<PackageReference Include="Microsoft.Toolkit.Uwp.UI.Controls">
<Version>2.0.0</Version>
<Version>2.2.0</Version>
</PackageReference>
</ItemGroup>
<PropertyGroup Condition=" '$(VisualStudioVersion)' == '' or '$(VisualStudioVersion)' &lt; '14.0' ">
@@ -148,7 +148,7 @@
</PivotItem>
<PivotItem Header="Log">
<StackPanel Background="{ThemeResource ApplicationPageBackgroundThemeBrush}">
<Button Click="Clear" Width="120">Clear</Button>
<Button Click="ClearLog" Width="120">Clear</Button>
</StackPanel>
</PivotItem>
</Pivot>
@@ -42,7 +42,8 @@ namespace MQTTnet.TestApp.UniversalWindows
foreach (var traceMessage in _traceMessages)
{
logText.AppendFormat(
"[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}", traceMessage.Timestamp,
"[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}",
traceMessage.Timestamp,
traceMessage.Level,
traceMessage.Source,
traceMessage.ThreadId,
@@ -197,8 +198,13 @@ namespace MQTTnet.TestApp.UniversalWindows
}
}
private void Clear(object sender, RoutedEventArgs e)
private void ClearLog(object sender, RoutedEventArgs e)
{
while (_traceMessages.Count > 0)
{
_traceMessages.TryDequeue(out _);
}
Trace.Text = string.Empty;
}