mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-15 00:15:15 +08:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
312773d054 | ||
|
|
1ee48a978f | ||
|
|
5ddb6b77e2 | ||
|
|
e0284886b5 | ||
|
|
44c07c2034 | ||
|
|
f6b38081fc | ||
|
|
c570509628 | ||
|
|
e98ff1d6d6 | ||
|
|
0f54663789 | ||
|
|
f91d935708 | ||
|
|
fb8b60fccc | ||
|
|
ad8506cc30 | ||
|
|
d6f88acb8d | ||
|
|
333fc4c387 | ||
|
|
5f9214e9a2 | ||
|
|
f974dcb647 | ||
|
|
2cb1c544f5 | ||
|
|
644e948489 | ||
|
|
d12d2648b7 | ||
|
|
0ee3c8e135 | ||
|
|
bc959a8c3d | ||
|
|
69781d822a | ||
|
|
611fabcd01 | ||
|
|
24ab285066 | ||
|
|
ee6ec14bf8 | ||
|
|
f9d67e2cd2 | ||
|
|
a72516fdcc | ||
|
|
ebfc6aab6b |
@@ -291,3 +291,5 @@ __pycache__/
|
||||
Build/nuget.exe
|
||||
*.js
|
||||
*.map
|
||||
|
||||
/Tests/MQTTnet.TestApp.NetCore/RetainedMessages.json
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
<package >
|
||||
<metadata>
|
||||
<id>MQTTnet.AspNetCore</id>
|
||||
<version>2.5.3</version>
|
||||
<version>2.6.0</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.5.3.
|
||||
<releaseNotes>* Updated to MQTTnet 2.6.0.
|
||||
</releaseNotes>
|
||||
<copyright>Copyright Christian Kratky 2016-2017</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.5.3-rc1" />
|
||||
<dependency id="MQTTnet" version="2.6.0" />
|
||||
</group>
|
||||
</dependencies>
|
||||
</metadata>
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
<?xml version="1.0"?>
|
||||
<package >
|
||||
<metadata>
|
||||
<id>MQTTnet.Extensions.Rpc</id>
|
||||
<version>2.6.0</version>
|
||||
<authors>Christian Kratky</authors>
|
||||
<owners>Christian Kratky</owners>
|
||||
<licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
|
||||
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
|
||||
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
|
||||
<requireLicenseAcceptance>false</requireLicenseAcceptance>
|
||||
<description>This is a extension library which allows executing synchronous device calls including a response using MQTTnet.</description>
|
||||
<releaseNotes>* Initial version.
|
||||
</releaseNotes>
|
||||
<copyright>Copyright Christian Kratky 2016-2017</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-rc1" />
|
||||
</group>
|
||||
<group targetFramework="netstandard1.3">
|
||||
<dependency id="MQTTnet" version="2.6.0-rc1" />
|
||||
</group>
|
||||
|
||||
<group targetFramework="netstandard2.0">
|
||||
<dependency id="MQTTnet" version="2.6.0-rc1" />
|
||||
</group>
|
||||
|
||||
<group targetFramework="uap10.0">
|
||||
<dependency id="MQTTnet" version="2.6.0-rc1" />
|
||||
</group>
|
||||
|
||||
<group targetFramework="net452">
|
||||
<dependency id="MQTTnet" version="2.6.0-rc1" />
|
||||
</group>
|
||||
|
||||
<group targetFramework="net461">
|
||||
<dependency id="MQTTnet" version="2.6.0-rc1" />
|
||||
</group>
|
||||
</dependencies>
|
||||
</metadata>
|
||||
|
||||
<files>
|
||||
<!-- .NET Standard 2.0 -->
|
||||
<file src="..\Frameworks\MQTTnet.AspNetCore\bin\Release\netstandard2.0\MQTTnet.AspNetCore.*" target="lib\netstandard2.0\"/>
|
||||
</files>
|
||||
</package>
|
||||
+22
-9
@@ -2,7 +2,7 @@
|
||||
<package >
|
||||
<metadata>
|
||||
<id>MQTTnet</id>
|
||||
<version>2.5.3</version>
|
||||
<version>2.6.0</version>
|
||||
<authors>Christian Kratky</authors>
|
||||
<owners>Christian Kratky</owners>
|
||||
<licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
|
||||
@@ -10,9 +10,26 @@
|
||||
<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] Removed all dependencies to other libraries (BREAKING CHANGE!).
|
||||
* [Core] Updated SDK libraries.
|
||||
* [Client] Fixed broken support for WebSocketSecure connections (Thanks to @StAI).
|
||||
<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>
|
||||
<copyright>Copyright Christian Kratky 2016-2017</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>
|
||||
@@ -46,23 +63,19 @@
|
||||
</metadata>
|
||||
|
||||
<files>
|
||||
|
||||
<!-- .NET Standard 1.3 -->
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard1.3\MQTTnet.Core.*" target="lib\netstandard1.3\"/>
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
|
||||
|
||||
<!-- .NET Standard 2.0 -->
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard2.0\MQTTnet.Core.*" target="lib\netstandard2.0\"/>
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard2.0\MQTTnet.*" target="lib\netstandard2.0\"/>
|
||||
|
||||
<!-- Universal Windows -->
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\uap10.0\MQTTnet.Core.*" target="lib\uap10.0\"/>
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\uap10.0\MQTTnet.*" target="lib\uap10.0\"/>
|
||||
|
||||
<!-- .NET Framework -->
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net452\MQTTnet.Core.*" target="lib\net452\"/>
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net452\MQTTnet.*" target="lib\net452\"/>
|
||||
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net461\MQTTnet.Core.*" target="lib\net461\"/>
|
||||
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net461\MQTTnet.*" target="lib\net461\"/>
|
||||
|
||||
</files>
|
||||
|
||||
@@ -8,16 +8,26 @@ $path = &$vswhere -latest -products * -requires Microsoft.Component.MSBuild -pro
|
||||
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
|
||||
|
||||
# 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
|
||||
|
||||
# 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
|
||||
&$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
|
||||
&$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
|
||||
&$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
|
||||
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
|
||||
.\NuGet.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>netstandard1.3;netstandard2.0;net452;net461;uap10.0</TargetFrameworks>
|
||||
<AssemblyVersion>0.0.0.0</AssemblyVersion>
|
||||
<FileVersion>0.0.0.0</FileVersion>
|
||||
<Product />
|
||||
<Company />
|
||||
<Authors />
|
||||
<PackageId />
|
||||
<Version>0.0.0.0</Version>
|
||||
</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>
|
||||
<LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
@@ -0,0 +1,101 @@
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Internal;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Extensions.Rpc
|
||||
{
|
||||
public sealed class MqttRpcClient : IDisposable
|
||||
{
|
||||
private const string ResponseTopic = "$MQTTnet.RPC/+/+/response";
|
||||
private readonly ConcurrentDictionary<string, TaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, TaskCompletionSource<byte[]>>();
|
||||
private readonly IMqttClient _mqttClient;
|
||||
private bool _isEnabled;
|
||||
|
||||
public MqttRpcClient(IMqttClient mqttClient)
|
||||
{
|
||||
_mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
|
||||
|
||||
_mqttClient.ApplicationMessageReceived += OnApplicationMessageReceived;
|
||||
}
|
||||
|
||||
public async Task EnableAsync()
|
||||
{
|
||||
await _mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(ResponseTopic).WithAtLeastOnceQoS().Build());
|
||||
_isEnabled = true;
|
||||
}
|
||||
|
||||
public async Task DisableAsync()
|
||||
{
|
||||
await _mqttClient.UnsubscribeAsync(ResponseTopic);
|
||||
_isEnabled = false;
|
||||
}
|
||||
|
||||
public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
|
||||
{
|
||||
if (methodName == null) throw new ArgumentNullException(nameof(methodName));
|
||||
|
||||
if (methodName.Contains("/") || methodName.Contains("+") || methodName.Contains("#"))
|
||||
{
|
||||
throw new ArgumentException("The method name cannot contain /, + or #.");
|
||||
}
|
||||
|
||||
if (!_isEnabled)
|
||||
{
|
||||
throw new InvalidOperationException("The RPC client is not enabled.");
|
||||
}
|
||||
|
||||
var requestTopic = $"$MQTTnet.RPC/{Guid.NewGuid():N}/{methodName}";
|
||||
var responseTopic = requestTopic + "/response";
|
||||
|
||||
var requestMessage = new MqttApplicationMessageBuilder()
|
||||
.WithTopic(requestTopic)
|
||||
.WithPayload(payload)
|
||||
.WithQualityOfServiceLevel(qualityOfServiceLevel)
|
||||
.Build();
|
||||
|
||||
try
|
||||
{
|
||||
var tcs = new TaskCompletionSource<byte[]>();
|
||||
if (!_waitingCalls.TryAdd(responseTopic, tcs))
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
}
|
||||
|
||||
await _mqttClient.PublishAsync(requestMessage);
|
||||
return await tcs.Task.TimeoutAfter(timeout);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_waitingCalls.TryRemove(responseTopic, out _);
|
||||
}
|
||||
}
|
||||
|
||||
private void OnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs eventArgs)
|
||||
{
|
||||
if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out TaskCompletionSource<byte[]> tcs))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (tcs.Task.IsCompleted || tcs.Task.IsCanceled)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
tcs.TrySetResult(eventArgs.ApplicationMessage.Payload);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
foreach (var tcs in _waitingCalls)
|
||||
{
|
||||
tcs.Value.SetCanceled();
|
||||
}
|
||||
|
||||
_waitingCalls.Clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
using System;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using MQTTnet.Core.Server;
|
||||
using System.Linq;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
|
||||
@@ -2,8 +2,17 @@
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>netstandard2.0</TargetFramework>
|
||||
<AssemblyVersion>2.5.2.0</AssemblyVersion>
|
||||
<FileVersion>2.5.2.0</FileVersion>
|
||||
<AssemblyVersion>0.0.0.0</AssemblyVersion>
|
||||
<FileVersion>0.0.0.0</FileVersion>
|
||||
<Version>0.0.0.0</Version>
|
||||
<Product />
|
||||
<Company />
|
||||
<Authors />
|
||||
<PackageId />
|
||||
</PropertyGroup>
|
||||
|
||||
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
|
||||
<DefineConstants>RELEASE;NETSTANDARD2_0</DefineConstants>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
|
||||
@@ -3,17 +3,17 @@ using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Server;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public class MqttHostedServer : MqttServer, IHostedService
|
||||
{
|
||||
private readonly MqttServerOptions _options;
|
||||
private readonly IMqttServerOptions _options;
|
||||
|
||||
public MqttHostedServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(adapters, logger)
|
||||
public MqttHostedServer(IMqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(adapters, logger)
|
||||
{
|
||||
_options = options ?? throw new ArgumentNullException(nameof(options));
|
||||
}
|
||||
|
||||
@@ -1,18 +1,18 @@
|
||||
using System;
|
||||
using System.Net.WebSockets;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Core.Server;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Serializer;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public class MqttWebSocketServerAdapter : IMqttServerAdapter, IDisposable
|
||||
public sealed class MqttWebSocketServerAdapter : IMqttServerAdapter, IDisposable
|
||||
{
|
||||
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
|
||||
public Task StartAsync(MqttServerOptions options)
|
||||
public Task StartAsync(IMqttServerOptions options)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ using System.IO;
|
||||
using System.Net.WebSockets;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Channel;
|
||||
using MQTTnet.Channel;
|
||||
using MQTTnet.Implementations;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
|
||||
@@ -1,15 +1,15 @@
|
||||
using System;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Server;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options)
|
||||
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, IMqttServerOptions options)
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
|
||||
|
||||
+3
-3
@@ -2,10 +2,10 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Packets;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.Core.Adapter
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public interface IMqttChannelAdapter
|
||||
{
|
||||
+3
-3
@@ -1,14 +1,14 @@
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Server;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.Core.Adapter
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public interface IMqttServerAdapter
|
||||
{
|
||||
event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
|
||||
Task StartAsync(MqttServerOptions options);
|
||||
Task StartAsync(IMqttServerOptions options);
|
||||
Task StopAsync();
|
||||
}
|
||||
}
|
||||
+58
-47
@@ -1,22 +1,25 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Net.Sockets;
|
||||
using System.Runtime.InteropServices;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Channel;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Core.Internal;
|
||||
using MQTTnet.Core.Packets;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Channel;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Internal;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.Core.Adapter
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public class MqttChannelAdapter : IMqttChannelAdapter
|
||||
{
|
||||
private const uint ErrorOperationAborted = 0x800703E3;
|
||||
|
||||
private static readonly byte[] EmptyBody = new byte[0];
|
||||
|
||||
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
|
||||
private readonly IMqttNetLogger _logger;
|
||||
private readonly IMqttChannel _channel;
|
||||
@@ -30,23 +33,23 @@ namespace MQTTnet.Core.Adapter
|
||||
|
||||
public IMqttPacketSerializer PacketSerializer { get; }
|
||||
|
||||
public async Task ConnectAsync(TimeSpan timeout)
|
||||
public Task ConnectAsync(TimeSpan timeout)
|
||||
{
|
||||
_logger.Info<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
|
||||
_logger.Trace<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
|
||||
|
||||
await ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
|
||||
return ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
|
||||
}
|
||||
|
||||
public async Task DisconnectAsync(TimeSpan timeout)
|
||||
public Task DisconnectAsync(TimeSpan timeout)
|
||||
{
|
||||
_logger.Info<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
|
||||
_logger.Trace<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
|
||||
|
||||
await ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
|
||||
return ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
|
||||
}
|
||||
|
||||
public async Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets)
|
||||
public Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets)
|
||||
{
|
||||
await ExecuteAndWrapExceptionAsync(async () =>
|
||||
return ExecuteAndWrapExceptionAsync(async () =>
|
||||
{
|
||||
await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
@@ -60,8 +63,11 @@ namespace MQTTnet.Core.Adapter
|
||||
|
||||
_logger.Trace<MqttChannelAdapter>("TX >>> {0} [Timeout={1}]", packet, timeout);
|
||||
|
||||
var writeBuffer = PacketSerializer.Serialize(packet);
|
||||
await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
|
||||
var chunks = PacketSerializer.Serialize(packet);
|
||||
foreach (var chunk in chunks)
|
||||
{
|
||||
await _channel.SendStream.WriteAsync(chunk.Array, chunk.Offset, chunk.Count, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
if (timeout > TimeSpan.Zero)
|
||||
@@ -85,36 +91,29 @@ namespace MQTTnet.Core.Adapter
|
||||
MqttBasePacket packet = null;
|
||||
await ExecuteAndWrapExceptionAsync(async () =>
|
||||
{
|
||||
ReceivedMqttPacket receivedMqttPacket = null;
|
||||
try
|
||||
ReceivedMqttPacket receivedMqttPacket;
|
||||
if (timeout > TimeSpan.Zero)
|
||||
{
|
||||
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();
|
||||
}
|
||||
|
||||
packet = PacketSerializer.Deserialize(receivedMqttPacket);
|
||||
if (packet == null)
|
||||
{
|
||||
throw new MqttProtocolViolationException("Received malformed packet.");
|
||||
}
|
||||
|
||||
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
|
||||
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
else
|
||||
{
|
||||
receivedMqttPacket?.Dispose();
|
||||
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
});
|
||||
|
||||
if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw new TaskCanceledException();
|
||||
}
|
||||
|
||||
packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
|
||||
if (packet == null)
|
||||
{
|
||||
throw new MqttProtocolViolationException("Received malformed packet.");
|
||||
}
|
||||
|
||||
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
|
||||
}).ConfigureAwait(false);
|
||||
|
||||
return packet;
|
||||
}
|
||||
@@ -129,7 +128,7 @@ namespace MQTTnet.Core.Adapter
|
||||
|
||||
if (header.BodyLength == 0)
|
||||
{
|
||||
return new ReceivedMqttPacket(header, new MemoryStream(0));
|
||||
return new ReceivedMqttPacket(header, EmptyBody);
|
||||
}
|
||||
|
||||
var body = new byte[header.BodyLength];
|
||||
@@ -140,8 +139,8 @@ namespace MQTTnet.Core.Adapter
|
||||
var readBytesCount = await stream.ReadAsync(body, offset, body.Length - offset, cancellationToken).ConfigureAwait(false);
|
||||
offset += readBytesCount;
|
||||
} while (offset < header.BodyLength);
|
||||
|
||||
return new ReceivedMqttPacket(header, new MemoryStream(body, 0, body.Length));
|
||||
|
||||
return new ReceivedMqttPacket(header, body);
|
||||
}
|
||||
|
||||
private static async Task ExecuteAndWrapExceptionAsync(Func<Task> action)
|
||||
@@ -168,13 +167,25 @@ namespace MQTTnet.Core.Adapter
|
||||
}
|
||||
catch (COMException comException)
|
||||
{
|
||||
if ((uint)comException.HResult == ErrorOperationAborted)
|
||||
if ((uint) comException.HResult == ErrorOperationAborted)
|
||||
{
|
||||
throw new OperationCanceledException();
|
||||
}
|
||||
|
||||
throw new MqttCommunicationException(comException);
|
||||
}
|
||||
catch (IOException exception)
|
||||
{
|
||||
if (exception.InnerException is SocketException socketException)
|
||||
{
|
||||
if (socketException.SocketErrorCode == SocketError.ConnectionAborted)
|
||||
{
|
||||
throw new OperationCanceledException();
|
||||
}
|
||||
}
|
||||
|
||||
throw new MqttCommunicationException(exception);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw new MqttCommunicationException(exception);
|
||||
+2
-2
@@ -1,9 +1,9 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Packets;
|
||||
using MQTTnet.Packets;
|
||||
|
||||
namespace MQTTnet.Core.Adapter
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public static class MqttChannelAdapterExtensions
|
||||
{
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Adapter
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public class MqttConnectingFailedException : MqttCommunicationException
|
||||
{
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.Core.Adapter
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public class MqttServerAdapterClientAcceptedEventArgs : EventArgs
|
||||
{
|
||||
@@ -0,0 +1,18 @@
|
||||
using System;
|
||||
using MQTTnet.Packets;
|
||||
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public class ReceivedMqttPacket
|
||||
{
|
||||
public ReceivedMqttPacket(MqttPacketHeader header, byte[] body)
|
||||
{
|
||||
Header = header ?? throw new ArgumentNullException(nameof(header));
|
||||
Body = body ?? throw new ArgumentNullException(nameof(body));
|
||||
}
|
||||
|
||||
public MqttPacketHeader Header { get; }
|
||||
|
||||
public byte[] Body { get; }
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.Core
|
||||
namespace MQTTnet
|
||||
{
|
||||
public static class ApplicationMessagePublisherExtensions
|
||||
{
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
using System.Threading.Tasks;
|
||||
using System.IO;
|
||||
using System.IO;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.Core.Channel
|
||||
namespace MQTTnet.Channel
|
||||
{
|
||||
public interface IMqttChannel
|
||||
{
|
||||
+2
-3
@@ -1,9 +1,8 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Packets;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
|
||||
{
|
||||
@@ -16,6 +15,6 @@ namespace MQTTnet.Core.Client
|
||||
Task DisconnectAsync();
|
||||
|
||||
Task<IList<MqttSubscribeResult>> SubscribeAsync(IEnumerable<TopicFilter> topicFilters);
|
||||
Task UnsubscribeAsync(IEnumerable<string> topicFilters);
|
||||
Task UnsubscribeAsync(IEnumerable<string> topics);
|
||||
}
|
||||
}
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public interface IMqttClientAdapterFactory
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public interface IMqttClientChannelOptions
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public interface IMqttClientCredentials
|
||||
{
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.ManagedClient;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.ManagedClient;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public interface IMqttClientFactory
|
||||
{
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public interface IMqttClientOptions
|
||||
{
|
||||
+36
-50
@@ -3,19 +3,17 @@ using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Core.Internal;
|
||||
using MQTTnet.Core.Packets;
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Internal;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClient : IMqttClient
|
||||
public class MqttClient : IMqttClient, IDisposable
|
||||
{
|
||||
private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
|
||||
|
||||
private readonly IMqttClientAdapterFactory _adapterFactory;
|
||||
private readonly MqttPacketDispatcher _packetDispatcher;
|
||||
private readonly IMqttNetLogger _logger;
|
||||
@@ -133,7 +131,7 @@ namespace MQTTnet.Core.Client
|
||||
TopicFilters = topicFilters.ToList()
|
||||
};
|
||||
|
||||
await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket);
|
||||
await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
|
||||
@@ -141,21 +139,21 @@ namespace MQTTnet.Core.Client
|
||||
ThrowIfNotConnected();
|
||||
|
||||
var publishPackets = applicationMessages.Select(m => m.ToPublishPacket());
|
||||
var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key);
|
||||
|
||||
foreach (var qosGroup in publishPackets.GroupBy(p => p.QualityOfServiceLevel))
|
||||
foreach (var qosGroup in packetGroups)
|
||||
{
|
||||
var qosPackets = qosGroup.ToArray();
|
||||
switch (qosGroup.Key)
|
||||
{
|
||||
case MqttQualityOfServiceLevel.AtMostOnce:
|
||||
{
|
||||
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
|
||||
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets).ConfigureAwait(false);
|
||||
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosGroup).ConfigureAwait(false);
|
||||
break;
|
||||
}
|
||||
case MqttQualityOfServiceLevel.AtLeastOnce:
|
||||
{
|
||||
foreach (var publishPacket in qosPackets)
|
||||
foreach (var publishPacket in qosGroup)
|
||||
{
|
||||
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
|
||||
await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
|
||||
@@ -165,7 +163,7 @@ namespace MQTTnet.Core.Client
|
||||
}
|
||||
case MqttQualityOfServiceLevel.ExactlyOnce:
|
||||
{
|
||||
foreach (var publishPacket in qosPackets)
|
||||
foreach (var publishPacket in qosGroup)
|
||||
{
|
||||
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
|
||||
var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
|
||||
@@ -182,6 +180,11 @@ namespace MQTTnet.Core.Client
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_cancellationTokenSource?.Dispose();
|
||||
}
|
||||
|
||||
private async Task<MqttConnAckPacket> AuthenticateAsync(MqttApplicationMessage willApplicationMessage)
|
||||
{
|
||||
var connectPacket = new MqttConnectPacket
|
||||
@@ -249,28 +252,28 @@ namespace MQTTnet.Core.Client
|
||||
try
|
||||
{
|
||||
_logger.Info<MqttClient>("Received <<< {0}", packet);
|
||||
|
||||
if (packet is MqttPublishPacket publishPacket)
|
||||
{
|
||||
await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (packet is MqttPingReqPacket)
|
||||
{
|
||||
await SendAsync(new MqttPingRespPacket());
|
||||
await SendAsync(new MqttPingRespPacket()).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (packet is MqttDisconnectPacket)
|
||||
{
|
||||
await DisconnectAsync();
|
||||
return;
|
||||
}
|
||||
|
||||
if (packet is MqttPublishPacket publishPacket)
|
||||
{
|
||||
await ProcessReceivedPublishPacket(publishPacket);
|
||||
await DisconnectAsync().ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (packet is MqttPubRelPacket pubRelPacket)
|
||||
{
|
||||
await ProcessReceivedPubRelPacket(pubRelPacket);
|
||||
await ProcessReceivedPubRelPacket(pubRelPacket).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -295,32 +298,25 @@ namespace MQTTnet.Core.Client
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
|
||||
private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket)
|
||||
{
|
||||
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
|
||||
{
|
||||
FireApplicationMessageReceivedEvent(publishPacket);
|
||||
return;
|
||||
return Task.FromResult(0);
|
||||
}
|
||||
|
||||
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
|
||||
{
|
||||
FireApplicationMessageReceivedEvent(publishPacket);
|
||||
await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
|
||||
return;
|
||||
return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
|
||||
}
|
||||
|
||||
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
|
||||
{
|
||||
// QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
|
||||
lock (_unacknowledgedPublishPackets)
|
||||
{
|
||||
_unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
|
||||
}
|
||||
|
||||
FireApplicationMessageReceivedEvent(publishPacket);
|
||||
await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
|
||||
return;
|
||||
return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
|
||||
}
|
||||
|
||||
throw new MqttCommunicationException("Received a not supported QoS level.");
|
||||
@@ -328,11 +324,6 @@ namespace MQTTnet.Core.Client
|
||||
|
||||
private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)
|
||||
{
|
||||
lock (_unacknowledgedPublishPackets)
|
||||
{
|
||||
_unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
|
||||
}
|
||||
|
||||
return SendAsync(pubRelPacket.CreateResponse<MqttPubCompPacket>());
|
||||
}
|
||||
|
||||
@@ -361,13 +352,8 @@ namespace MQTTnet.Core.Client
|
||||
{
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false);
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
|
||||
await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
@@ -454,7 +440,7 @@ namespace MQTTnet.Core.Client
|
||||
{
|
||||
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
|
||||
Task.Run(
|
||||
() => ProcessReceivedPacketAsync(packet),
|
||||
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
|
||||
}
|
||||
@@ -465,7 +451,7 @@ namespace MQTTnet.Core.Client
|
||||
|
||||
#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),
|
||||
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
|
||||
|
||||
@@ -479,7 +465,7 @@ namespace MQTTnet.Core.Client
|
||||
{
|
||||
#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),
|
||||
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
|
||||
}
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientConnectResult
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientConnectedEventArgs : EventArgs
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientCredentials : IMqttClientCredentials
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientDisconnectedEventArgs : EventArgs
|
||||
{
|
||||
+10
-1
@@ -2,8 +2,9 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public static class MqttClientExtensions
|
||||
{
|
||||
@@ -15,6 +16,14 @@ namespace MQTTnet.Core.Client
|
||||
return client.SubscribeAsync(topicFilters.ToList());
|
||||
}
|
||||
|
||||
public static Task<IList<MqttSubscribeResult>> SubscribeAsync(this IMqttClient client, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
|
||||
{
|
||||
if (client == null) throw new ArgumentNullException(nameof(client));
|
||||
if (topic == null) throw new ArgumentNullException(nameof(topic));
|
||||
|
||||
return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
|
||||
}
|
||||
|
||||
public static Task UnsubscribeAsync(this IMqttClient client, params string[] topicFilters)
|
||||
{
|
||||
if (client == null) throw new ArgumentNullException(nameof(client));
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientOptions : IMqttClientOptions
|
||||
{
|
||||
+5
-5
@@ -1,8 +1,8 @@
|
||||
using System;
|
||||
using System.Linq;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientOptionsBuilder
|
||||
{
|
||||
@@ -119,11 +119,11 @@ namespace MQTTnet.Core.Client
|
||||
|
||||
if (_tcpOptions != null)
|
||||
{
|
||||
_options.ChannelOptions = _tcpOptions;
|
||||
_tcpOptions.TlsOptions = _tlsOptions;
|
||||
}
|
||||
else
|
||||
else if (_webSocketOptions != null)
|
||||
{
|
||||
_options.ChannelOptions = _webSocketOptions;
|
||||
_webSocketOptions.TlsOptions = _tlsOptions;
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientTcpOptions : IMqttClientChannelOptions
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public static class MqttClientTcpOptionsExtensions
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientTlsOptions
|
||||
{
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Net;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClientWebSocketOptions : IMqttClientChannelOptions
|
||||
{
|
||||
+5
-5
@@ -1,12 +1,12 @@
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Core.Internal;
|
||||
using MQTTnet.Core.Packets;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Internal;
|
||||
using MQTTnet.Packets;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttPacketDispatcher
|
||||
{
|
||||
+2
-3
@@ -1,7 +1,6 @@
|
||||
using MQTTnet.Core.Packets;
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Client
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttSubscribeResult
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Diagnostics
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public interface IMqttNetLogger
|
||||
{
|
||||
@@ -1,50 +0,0 @@
|
||||
using System;
|
||||
using System.Text;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public static class MqttNetConsoleLogger
|
||||
{
|
||||
private static readonly object Lock = new object();
|
||||
|
||||
public static void ForwardToConsole()
|
||||
{
|
||||
MqttNetGlobalLog.LogMessagePublished -= PrintToConsole;
|
||||
MqttNetGlobalLog.LogMessagePublished += PrintToConsole;
|
||||
}
|
||||
|
||||
private static void PrintToConsole(object sender, MqttNetLogMessagePublishedEventArgs e)
|
||||
{
|
||||
var output = new StringBuilder();
|
||||
output.AppendLine($">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}");
|
||||
if (e.TraceMessage.Exception != null)
|
||||
{
|
||||
output.AppendLine(e.TraceMessage.Exception.ToString());
|
||||
}
|
||||
|
||||
lock (Lock)
|
||||
{
|
||||
var backupColor = Console.ForegroundColor;
|
||||
switch (e.TraceMessage.Level)
|
||||
{
|
||||
case MqttNetLogLevel.Error:
|
||||
Console.ForegroundColor = ConsoleColor.Red;
|
||||
break;
|
||||
case MqttNetLogLevel.Warning:
|
||||
Console.ForegroundColor = ConsoleColor.Yellow;
|
||||
break;
|
||||
case MqttNetLogLevel.Info:
|
||||
Console.ForegroundColor = ConsoleColor.Green;
|
||||
break;
|
||||
case MqttNetLogLevel.Verbose:
|
||||
Console.ForegroundColor = ConsoleColor.Gray;
|
||||
break;
|
||||
}
|
||||
|
||||
Console.Write(output);
|
||||
Console.ForegroundColor = backupColor;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,8 +1,8 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Diagnostics
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public static class MqttNetGlobalLog
|
||||
public static class MqttNetGlobalLogger
|
||||
{
|
||||
public static event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Diagnostics
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public enum MqttNetLogLevel
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Diagnostics
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public sealed class MqttNetLogMessage
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Diagnostics
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public sealed class MqttNetLogMessagePublishedEventArgs : EventArgs
|
||||
{
|
||||
+4
-4
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Diagnostics
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public class MqttNetLogger : IMqttNetLogger
|
||||
{
|
||||
@@ -10,7 +10,7 @@ namespace MQTTnet.Core.Diagnostics
|
||||
{
|
||||
_logId = logId;
|
||||
}
|
||||
|
||||
|
||||
public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
|
||||
|
||||
public void Trace<TSource>(string message, params object[] parameters)
|
||||
@@ -46,7 +46,7 @@ namespace MQTTnet.Core.Diagnostics
|
||||
private void Publish<TSource>(MqttNetLogLevel logLevel, Exception exception, string message, object[] parameters)
|
||||
{
|
||||
var hasLocalListeners = LogMessagePublished != null;
|
||||
var hasGlobalListeners = MqttNetGlobalLog.HasListeners;
|
||||
var hasGlobalListeners = MqttNetGlobalLogger.HasListeners;
|
||||
|
||||
if (!hasLocalListeners && !hasGlobalListeners)
|
||||
{
|
||||
@@ -62,7 +62,7 @@ namespace MQTTnet.Core.Diagnostics
|
||||
|
||||
if (hasGlobalListeners)
|
||||
{
|
||||
MqttNetGlobalLog.Publish(traceMessage);
|
||||
MqttNetGlobalLogger.Publish(traceMessage);
|
||||
}
|
||||
|
||||
if (hasLocalListeners)
|
||||
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet
|
||||
namespace MQTTnet.Diagnostics
|
||||
{
|
||||
public static class TargetFrameworkInfoProvider
|
||||
{
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Exceptions
|
||||
namespace MQTTnet.Exceptions
|
||||
{
|
||||
public class MqttCommunicationException : Exception
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Exceptions
|
||||
namespace MQTTnet.Exceptions
|
||||
{
|
||||
public sealed class MqttCommunicationTimedOutException : MqttCommunicationException
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Exceptions
|
||||
namespace MQTTnet.Exceptions
|
||||
{
|
||||
public sealed class MqttProtocolViolationException : Exception
|
||||
{
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.Core
|
||||
namespace MQTTnet
|
||||
{
|
||||
public interface IApplicationMessagePublisher
|
||||
{
|
||||
+1
-2
@@ -1,7 +1,6 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Client;
|
||||
|
||||
namespace MQTTnet.Core
|
||||
namespace MQTTnet
|
||||
{
|
||||
public interface IApplicationMessageReceiver
|
||||
{
|
||||
@@ -1,8 +1,8 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
#if WINDOWS_UWP
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Server;
|
||||
using Windows.Networking.Sockets;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Serializer;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
@@ -21,7 +21,7 @@ namespace MQTTnet.Implementations
|
||||
|
||||
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
|
||||
public async Task StartAsync(MqttServerOptions options)
|
||||
public async Task StartAsync(IMqttServerOptions options)
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
|
||||
|
||||
@@ -7,10 +7,10 @@ using System.Security.Authentication;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Server;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Serializer;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
@@ -30,7 +30,7 @@ namespace MQTTnet.Implementations
|
||||
|
||||
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
|
||||
public Task StartAsync(MqttServerOptions options)
|
||||
public Task StartAsync(IMqttServerOptions options)
|
||||
{
|
||||
if (_cancellationTokenSource != null) throw new InvalidOperationException("Server is already started.");
|
||||
|
||||
@@ -39,10 +39,10 @@ namespace MQTTnet.Implementations
|
||||
if (options.DefaultEndpointOptions.IsEnabled)
|
||||
{
|
||||
_defaultEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
|
||||
_defaultEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetDefaultEndpointPort()));
|
||||
_defaultEndpointSocket.Bind(new IPEndPoint(options.DefaultEndpointOptions.BoundIPAddress, options.GetDefaultEndpointPort()));
|
||||
_defaultEndpointSocket.Listen(options.ConnectionBacklog);
|
||||
|
||||
Task.Run(() => AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
|
||||
Task.Run(async () => await AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
if (options.TlsEndpointOptions.IsEnabled)
|
||||
@@ -59,10 +59,10 @@ namespace MQTTnet.Implementations
|
||||
}
|
||||
|
||||
_tlsEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
|
||||
_tlsEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetTlsEndpointPort()));
|
||||
_tlsEndpointSocket.Bind(new IPEndPoint(options.TlsEndpointOptions.BoundIPAddress, options.GetTlsEndpointPort()));
|
||||
_tlsEndpointSocket.Listen(options.ConnectionBacklog);
|
||||
|
||||
Task.Run(() => AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
|
||||
Task.Run(async () => await AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return Task.FromResult(0);
|
||||
@@ -102,13 +102,29 @@ namespace MQTTnet.Implementations
|
||||
#else
|
||||
var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
|
||||
#endif
|
||||
|
||||
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger);
|
||||
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// 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)
|
||||
{
|
||||
if (exception.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);
|
||||
}
|
||||
@@ -129,14 +145,29 @@ namespace MQTTnet.Implementations
|
||||
|
||||
var sslStream = new SslStream(new NetworkStream(clientSocket));
|
||||
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
|
||||
|
||||
|
||||
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger);
|
||||
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// 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)
|
||||
{
|
||||
if (exception.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);
|
||||
}
|
||||
|
||||
@@ -8,8 +8,8 @@ using System.Threading.Tasks;
|
||||
using Windows.Networking;
|
||||
using Windows.Networking.Sockets;
|
||||
using Windows.Security.Cryptography.Certificates;
|
||||
using MQTTnet.Core.Channel;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Channel;
|
||||
using MQTTnet.Client;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
|
||||
@@ -5,10 +5,10 @@ using System.Net.Sockets;
|
||||
using System.Security.Authentication;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Channel;
|
||||
using MQTTnet.Core.Client;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using MQTTnet.Channel;
|
||||
using MQTTnet.Client;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
@@ -70,7 +70,7 @@ namespace MQTTnet.Implementations
|
||||
_sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
|
||||
await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(_options), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
|
||||
CreateStreams(_socket, _sslStream);
|
||||
}
|
||||
|
||||
@@ -130,7 +130,7 @@ namespace MQTTnet.Implementations
|
||||
|
||||
foreach (var certificate in options.TlsOptions.Certificates)
|
||||
{
|
||||
certificates.Add(new X509Certificate(certificate));
|
||||
certificates.Add(new X509Certificate2(certificate));
|
||||
}
|
||||
|
||||
return certificates;
|
||||
@@ -139,10 +139,7 @@ namespace MQTTnet.Implementations
|
||||
private void CreateStreams(Socket socket, Stream sslStream)
|
||||
{
|
||||
var stream = sslStream ?? new NetworkStream(socket);
|
||||
|
||||
//cannot use this as default buffering prevents from receiving the first connect message
|
||||
//need two streams otherwise read and write have to be synchronized
|
||||
|
||||
|
||||
//todo: if branch can be used with min dependency NetStandard1.6
|
||||
#if NET452 || NET461
|
||||
SendStream = new BufferedStream(stream, BufferSize);
|
||||
@@ -155,4 +152,4 @@ namespace MQTTnet.Implementations
|
||||
|
||||
}
|
||||
}
|
||||
#endif
|
||||
#endif
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
using MQTTnet.Core.Channel;
|
||||
using MQTTnet.Core.Client;
|
||||
using System;
|
||||
using System;
|
||||
using System.IO;
|
||||
using System.Net.WebSockets;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Channel;
|
||||
using MQTTnet.Client;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
|
||||
@@ -5,7 +5,7 @@ using System.Linq;
|
||||
using System.Net.WebSockets;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Exceptions;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MQTTnet.Core.Packets;
|
||||
using MQTTnet.Packets;
|
||||
|
||||
namespace MQTTnet.Core.Internal
|
||||
namespace MQTTnet.Internal
|
||||
{
|
||||
internal static class MqttApplicationMessageExtensions
|
||||
{
|
||||
+13
-9
@@ -1,21 +1,23 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Exceptions;
|
||||
|
||||
namespace MQTTnet.Core.Internal
|
||||
namespace MQTTnet.Internal
|
||||
{
|
||||
public static class TaskExtensions
|
||||
{
|
||||
public static async Task TimeoutAfter(this Task task, TimeSpan timeout)
|
||||
{
|
||||
using (var cancellationTokenSource = new CancellationTokenSource())
|
||||
if (task == null) throw new ArgumentNullException(nameof(task));
|
||||
|
||||
using (var timeoutCts = new CancellationTokenSource())
|
||||
{
|
||||
try
|
||||
{
|
||||
var timeoutTask = Task.Delay(timeout, cancellationTokenSource.Token);
|
||||
var timeoutTask = Task.Delay(timeout, timeoutCts.Token);
|
||||
var finishedTask = await Task.WhenAny(timeoutTask, task).ConfigureAwait(false);
|
||||
|
||||
|
||||
if (finishedTask == timeoutTask)
|
||||
{
|
||||
throw new MqttCommunicationTimedOutException();
|
||||
@@ -33,18 +35,20 @@ namespace MQTTnet.Core.Internal
|
||||
}
|
||||
finally
|
||||
{
|
||||
cancellationTokenSource.Cancel();
|
||||
timeoutCts.Cancel();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static async Task<TResult> TimeoutAfter<TResult>(this Task<TResult> task, TimeSpan timeout)
|
||||
{
|
||||
using (var cancellationTokenSource = new CancellationTokenSource())
|
||||
if (task == null) throw new ArgumentNullException(nameof(task));
|
||||
|
||||
using (var timeoutCts = new CancellationTokenSource())
|
||||
{
|
||||
try
|
||||
{
|
||||
var timeoutTask = Task.Delay(timeout, cancellationTokenSource.Token);
|
||||
var timeoutTask = Task.Delay(timeout, timeoutCts.Token);
|
||||
var finishedTask = await Task.WhenAny(timeoutTask, task).ConfigureAwait(false);
|
||||
|
||||
if (finishedTask == timeoutTask)
|
||||
@@ -66,7 +70,7 @@ namespace MQTTnet.Core.Internal
|
||||
}
|
||||
finally
|
||||
{
|
||||
cancellationTokenSource.Cancel();
|
||||
timeoutCts.Cancel();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,19 +1,23 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>netstandard1.3;netstandard2.0;net452;net461;uap10.0</TargetFrameworks>
|
||||
<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>2.5.3.0</AssemblyVersion>
|
||||
<FileVersion>2.5.3.0</FileVersion>
|
||||
<AssemblyVersion>0.0.0.0</AssemblyVersion>
|
||||
<FileVersion>0.0.0.0</FileVersion>
|
||||
<Version>0.0.0.0</Version>
|
||||
<Company />
|
||||
<Product />
|
||||
<Description />
|
||||
<Authors />
|
||||
<PackageId />
|
||||
<SignAssembly>true</SignAssembly>
|
||||
<AssemblyOriginatorKeyFile>codeSigningKey.pfx</AssemblyOriginatorKeyFile>
|
||||
<DelaySign>false</DelaySign>
|
||||
</PropertyGroup>
|
||||
|
||||
<PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
|
||||
@@ -26,16 +30,15 @@
|
||||
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
|
||||
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
|
||||
<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'" />
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
|
||||
</ItemGroup>
|
||||
<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" />
|
||||
|
||||
+3
-3
@@ -1,9 +1,9 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Client;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
|
||||
{
|
||||
@@ -16,6 +16,6 @@ namespace MQTTnet.Core.ManagedClient
|
||||
Task StopAsync();
|
||||
|
||||
Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters);
|
||||
Task UnsubscribeAsync(IEnumerable<TopicFilter> topicFilters);
|
||||
Task UnsubscribeAsync(IEnumerable<string> topics);
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Client;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public interface IManagedMqttClientOptions
|
||||
{
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public interface IManagedMqttClientStorage
|
||||
{
|
||||
+133
-69
@@ -4,18 +4,18 @@ using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public class ManagedMqttClient : IManagedMqttClient
|
||||
public class ManagedMqttClient : IManagedMqttClient, IDisposable
|
||||
{
|
||||
private readonly ManagedMqttClientStorageManager _storageManager = new ManagedMqttClientStorageManager();
|
||||
private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
|
||||
private readonly HashSet<TopicFilter> _subscriptions = new HashSet<TopicFilter>();
|
||||
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
|
||||
private readonly SemaphoreSlim _subscriptionsSemaphore = new SemaphoreSlim(1, 1);
|
||||
|
||||
private readonly IMqttClient _mqttClient;
|
||||
private readonly IMqttNetLogger _logger;
|
||||
@@ -23,7 +23,9 @@ namespace MQTTnet.Core.ManagedClient
|
||||
private CancellationTokenSource _connectionCancellationToken;
|
||||
private CancellationTokenSource _publishingCancellationToken;
|
||||
|
||||
private ManagedMqttClientStorageManager _storageManager;
|
||||
private IManagedMqttClientOptions _options;
|
||||
|
||||
private bool _subscriptionsNotPushed;
|
||||
|
||||
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
|
||||
@@ -55,15 +57,11 @@ namespace MQTTnet.Core.ManagedClient
|
||||
if (_connectionCancellationToken != null) throw new InvalidOperationException("The managed client is already started.");
|
||||
|
||||
_options = options;
|
||||
await _storageManager.SetStorageAsync(_options.Storage).ConfigureAwait(false);
|
||||
|
||||
|
||||
if (_options.Storage != null)
|
||||
{
|
||||
var loadedMessages = await _options.Storage.LoadQueuedMessagesAsync().ConfigureAwait(false);
|
||||
foreach (var loadedMessage in loadedMessages)
|
||||
{
|
||||
_messageQueue.Add(loadedMessage);
|
||||
}
|
||||
_storageManager = new ManagedMqttClientStorageManager(_options.Storage);
|
||||
await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_connectionCancellationToken = new CancellationTokenSource();
|
||||
@@ -77,11 +75,8 @@ namespace MQTTnet.Core.ManagedClient
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
_connectionCancellationToken?.Cancel(false);
|
||||
_connectionCancellationToken = null;
|
||||
|
||||
_publishingCancellationToken?.Cancel(false);
|
||||
_publishingCancellationToken = null;
|
||||
StopPublishing();
|
||||
StopMaintainingConnection();
|
||||
|
||||
while (_messageQueue.Any())
|
||||
{
|
||||
@@ -90,50 +85,66 @@ namespace MQTTnet.Core.ManagedClient
|
||||
|
||||
return Task.FromResult(0);
|
||||
}
|
||||
|
||||
|
||||
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
|
||||
{
|
||||
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
|
||||
|
||||
foreach (var applicationMessage in applicationMessages)
|
||||
{
|
||||
await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
|
||||
if (_storageManager != null)
|
||||
{
|
||||
await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_messageQueue.Add(applicationMessage);
|
||||
}
|
||||
}
|
||||
|
||||
public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
|
||||
public async Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
|
||||
{
|
||||
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
|
||||
|
||||
lock (_subscriptions)
|
||||
await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
foreach (var topicFilter in topicFilters)
|
||||
{
|
||||
if (_subscriptions.Add(topicFilter))
|
||||
{
|
||||
_subscriptionsNotPushed = true;
|
||||
}
|
||||
_subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
|
||||
_subscriptionsNotPushed = true;
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult(0);
|
||||
finally
|
||||
{
|
||||
_subscriptionsSemaphore.Release();
|
||||
}
|
||||
}
|
||||
|
||||
public Task UnsubscribeAsync(IEnumerable<TopicFilter> topicFilters)
|
||||
public async Task UnsubscribeAsync(IEnumerable<string> topics)
|
||||
{
|
||||
lock (_subscriptions)
|
||||
await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
foreach (var topicFilter in topicFilters)
|
||||
foreach (var topic in topics)
|
||||
{
|
||||
if (_subscriptions.Remove(topicFilter))
|
||||
if (_subscriptions.Remove(topic))
|
||||
{
|
||||
_subscriptionsNotPushed = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_subscriptionsSemaphore.Release();
|
||||
}
|
||||
}
|
||||
|
||||
return Task.FromResult(0);
|
||||
public void Dispose()
|
||||
{
|
||||
_messageQueue?.Dispose();
|
||||
_subscriptionsSemaphore?.Dispose();
|
||||
_connectionCancellationToken?.Dispose();
|
||||
_publishingCancellationToken?.Dispose();
|
||||
}
|
||||
|
||||
private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
|
||||
@@ -142,33 +153,48 @@ namespace MQTTnet.Core.ManagedClient
|
||||
{
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false);
|
||||
if (connectionState == ReconnectionResult.NotConnected)
|
||||
{
|
||||
_publishingCancellationToken?.Cancel(false);
|
||||
_publishingCancellationToken = null;
|
||||
await TryMaintainConnectionAsync(cancellationToken);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while maintaining connection.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await _mqttClient.DisconnectAsync().ConfigureAwait(false);
|
||||
_logger.Info<ManagedMqttClient>("Stopped");
|
||||
}
|
||||
}
|
||||
|
||||
await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
|
||||
continue;
|
||||
}
|
||||
private async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false);
|
||||
if (connectionState == ReconnectionResult.NotConnected)
|
||||
{
|
||||
StopPublishing();
|
||||
await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
|
||||
{
|
||||
await PushSubscriptionsAsync();
|
||||
if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
|
||||
{
|
||||
await PushSubscriptionsAsync().ConfigureAwait(false);
|
||||
|
||||
_publishingCancellationToken = new CancellationTokenSource();
|
||||
StartPublishing();
|
||||
|
||||
#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 PublishQueuedMessagesAsync(_publishingCancellationToken.Token), _publishingCancellationToken.Token).ConfigureAwait(false);
|
||||
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
if (connectionState == ReconnectionResult.StillConnected)
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(1), _connectionCancellationToken.Token).ConfigureAwait(false);
|
||||
}
|
||||
if (connectionState == ReconnectionResult.StillConnected)
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
@@ -182,13 +208,8 @@ namespace MQTTnet.Core.ManagedClient
|
||||
{
|
||||
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while maintaining connection.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await _mqttClient.DisconnectAsync().ConfigureAwait(false);
|
||||
_logger.Info<ManagedMqttClient>("Stopped");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
@@ -206,16 +227,19 @@ namespace MQTTnet.Core.ManagedClient
|
||||
continue;
|
||||
}
|
||||
|
||||
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
|
||||
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
|
||||
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while publishing queued application messages.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
_logger.Info<ManagedMqttClient>("Stopped publishing messages");
|
||||
_logger.Trace<ManagedMqttClient>("Stopped publishing messages.");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -224,6 +248,11 @@ namespace MQTTnet.Core.ManagedClient
|
||||
try
|
||||
{
|
||||
await _mqttClient.PublishAsync(message).ConfigureAwait(false);
|
||||
|
||||
if (_storageManager != null)
|
||||
{
|
||||
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (MqttCommunicationException exception)
|
||||
{
|
||||
@@ -245,13 +274,18 @@ namespace MQTTnet.Core.ManagedClient
|
||||
_logger.Info<ManagedMqttClient>(nameof(ManagedMqttClient), "Synchronizing subscriptions");
|
||||
|
||||
List<TopicFilter> subscriptions;
|
||||
lock (_subscriptions)
|
||||
await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
subscriptions = _subscriptions.ToList();
|
||||
subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList();
|
||||
_subscriptionsNotPushed = false;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_subscriptionsSemaphore.Release();
|
||||
}
|
||||
|
||||
if (!_subscriptions.Any())
|
||||
if (!subscriptions.Any())
|
||||
{
|
||||
return;
|
||||
}
|
||||
@@ -299,5 +333,35 @@ namespace MQTTnet.Core.ManagedClient
|
||||
{
|
||||
Connected?.Invoke(this, eventArgs);
|
||||
}
|
||||
|
||||
private void StartPublishing()
|
||||
{
|
||||
if (_publishingCancellationToken != null)
|
||||
{
|
||||
StopPublishing();
|
||||
}
|
||||
|
||||
var cts = new CancellationTokenSource();
|
||||
|
||||
_publishingCancellationToken = cts;
|
||||
|
||||
#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 PublishQueuedMessagesAsync(cts.Token).ConfigureAwait(false), cts.Token).ConfigureAwait(false);
|
||||
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
|
||||
}
|
||||
|
||||
private void StopPublishing()
|
||||
{
|
||||
_publishingCancellationToken?.Cancel(false);
|
||||
_publishingCancellationToken?.Dispose();
|
||||
_publishingCancellationToken = null;
|
||||
}
|
||||
|
||||
private void StopMaintainingConnection()
|
||||
{
|
||||
_connectionCancellationToken?.Cancel(false);
|
||||
_connectionCancellationToken?.Dispose();
|
||||
_connectionCancellationToken = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
+11
-2
@@ -1,7 +1,8 @@
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public static class ManagedMqttClientExtensions
|
||||
{
|
||||
@@ -12,7 +13,15 @@ namespace MQTTnet.Core.ManagedClient
|
||||
return managedClient.SubscribeAsync(topicFilters);
|
||||
}
|
||||
|
||||
public static Task UnsubscribeAsync(this IManagedMqttClient managedClient, params TopicFilter[] topicFilters)
|
||||
public static Task SubscribeAsync(this IManagedMqttClient managedClient, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
|
||||
{
|
||||
if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
|
||||
if (topic == null) throw new ArgumentNullException(nameof(topic));
|
||||
|
||||
return managedClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
|
||||
}
|
||||
|
||||
public static Task UnsubscribeAsync(this IManagedMqttClient managedClient, params string[] topicFilters)
|
||||
{
|
||||
if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
|
||||
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Client;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public class ManagedMqttClientOptions : IManagedMqttClientOptions
|
||||
{
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Client;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public class ManagedMqttClientOptionsBuilder
|
||||
{
|
||||
+13
-21
@@ -1,25 +1,27 @@
|
||||
using System.Collections.Generic;
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public class ManagedMqttClientStorageManager
|
||||
{
|
||||
private readonly List<MqttApplicationMessage> _applicationMessages = new List<MqttApplicationMessage>();
|
||||
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
|
||||
private IManagedMqttClientStorage _storage;
|
||||
private readonly IManagedMqttClientStorage _storage;
|
||||
|
||||
public async Task SetStorageAsync(IManagedMqttClientStorage storage)
|
||||
public ManagedMqttClientStorageManager(IManagedMqttClientStorage storage)
|
||||
{
|
||||
await _semaphore.WaitAsync().ConfigureAwait(false);
|
||||
try
|
||||
_storage = storage ?? throw new ArgumentNullException(nameof(storage));
|
||||
}
|
||||
|
||||
public async Task LoadQueuedMessagesAsync()
|
||||
{
|
||||
var loadedMessages = await _storage.LoadQueuedMessagesAsync().ConfigureAwait(false);
|
||||
foreach (var loadedMessage in loadedMessages)
|
||||
{
|
||||
_storage = storage;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_semaphore.Release();
|
||||
_applicationMessages.Add(loadedMessage);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,11 +30,6 @@ namespace MQTTnet.Core.ManagedClient
|
||||
await _semaphore.WaitAsync().ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
if (_storage == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_applicationMessages.Add(applicationMessage);
|
||||
await SaveAsync().ConfigureAwait(false);
|
||||
}
|
||||
@@ -47,11 +44,6 @@ namespace MQTTnet.Core.ManagedClient
|
||||
await _semaphore.WaitAsync().ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
if (_storage == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var index = _applicationMessages.IndexOf(applicationMessage);
|
||||
if (index == -1)
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.ManagedClient
|
||||
namespace MQTTnet.ManagedClient
|
||||
{
|
||||
public enum ReconnectionResult
|
||||
{
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using System;
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core
|
||||
namespace MQTTnet
|
||||
{
|
||||
public sealed class MqttApplicationMessage
|
||||
{
|
||||
+3
-3
@@ -2,10 +2,10 @@
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using MQTTnet.Core.Exceptions;
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core
|
||||
namespace MQTTnet
|
||||
{
|
||||
public class MqttApplicationMessageBuilder
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core
|
||||
namespace MQTTnet
|
||||
{
|
||||
public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
|
||||
{
|
||||
@@ -1,11 +1,11 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Client;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Implementations;
|
||||
using MQTTnet.Core.ManagedClient;
|
||||
using MQTTnet.Core.Server;
|
||||
using MQTTnet.Core.Diagnostics;
|
||||
using MQTTnet.ManagedClient;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet
|
||||
{
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public interface IMqttPacketWithIdentifier
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public abstract class MqttBasePacket
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier
|
||||
{
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttConnAckPacket : MqttBasePacket
|
||||
{
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MQTTnet.Core.Serializer;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttConnectPacket : MqttBasePacket
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttDisconnectPacket : MqttBasePacket
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public static class MqttPacketExtensions
|
||||
{
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public class MqttPacketHeader
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttPingReqPacket : MqttBasePacket
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttPingRespPacket : MqttBasePacket
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttPubAckPacket : MqttBasePublishPacket
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttPubCompPacket : MqttBasePublishPacket
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttPubRecPacket : MqttBasePublishPacket
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttPubRelPacket : MqttBasePublishPacket
|
||||
{
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttPublishPacket : MqttBasePublishPacket
|
||||
{
|
||||
+2
-2
@@ -1,8 +1,8 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using MQTTnet.Core.Protocol;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
|
||||
{
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttUnsubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
|
||||
{
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace MQTTnet.Core.Packets
|
||||
namespace MQTTnet.Packets
|
||||
{
|
||||
public sealed class MqttUnsubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Protocol
|
||||
namespace MQTTnet.Protocol
|
||||
{
|
||||
public enum MqttConnectReturnCode
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MQTTnet.Core.Protocol
|
||||
namespace MQTTnet.Protocol
|
||||
{
|
||||
public enum MqttControlPacketType
|
||||
{
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user