diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec
index 4245077b..c9936f35 100644
--- a/Build/MQTTnet.AspNetCore.nuspec
+++ b/Build/MQTTnet.AspNetCore.nuspec
@@ -2,7 +2,7 @@
MQTTnet.AspNetCore
- 2.7.5
+ 0.0.0
Christian Kratky
Christian Kratky
https://github.com/chkr1011/MQTTnet/blob/master/LICENSE
@@ -10,19 +10,20 @@
https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png
false
This is a support library to integrate MQTTnet into AspNetCore.
- * Updated to MQTTnet 2.7.5.
+ * Updated to MQTTnet 2.8.0.
Copyright Christian Kratky 2016-2018
- MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin
+ 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
-
-
-
+
+
+
+
-
+
-
\ No newline at end of file
+
diff --git a/Build/MQTTnet.Extensions.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec
new file mode 100644
index 00000000..ba5333f2
--- /dev/null
+++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec
@@ -0,0 +1,38 @@
+
+
+
+ MQTTnet.Extensions.ManagedClient
+ 0.0.0
+ Christian Kratky
+ Christian Kratky
+ https://github.com/chkr1011/MQTTnet/blob/master/LICENSE
+ https://github.com/chkr1011/MQTTnet
+ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png
+ false
+ This is an extension library which provides a managed MQTT client with additional features using MQTTnet.
+ * Updated to MQTTnet 2.8.0.
+
+ Copyright Christian Kratky 2016-2018
+ MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec
index cb31c21b..0df8e7e6 100644
--- a/Build/MQTTnet.Extensions.Rpc.nuspec
+++ b/Build/MQTTnet.Extensions.Rpc.nuspec
@@ -2,46 +2,37 @@
MQTTnet.Extensions.Rpc
- 2.7.5
+ 0.0.0
Christian Kratky
Christian Kratky
https://github.com/chkr1011/MQTTnet/blob/master/LICENSE
https://github.com/chkr1011/MQTTnet
https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png
false
- This is a extension library which allows executing synchronous device calls including a response using MQTTnet.
- * Updated to MQTTnet 2.7.5.
+ This is an extension library which allows executing synchronous device calls including a response using MQTTnet.
+ * Updated to MQTTnet 2.8.0.
- Copyright Christian Kratky 2016-2017
+ Copyright Christian Kratky 2016-2018
MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
-
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec
index 764a36c5..5cdf3d52 100644
--- a/Build/MQTTnet.nuspec
+++ b/Build/MQTTnet.nuspec
@@ -2,7 +2,7 @@
MQTTnet
- 2.8.0
+ 0.0.0
Christian Kratky
Christian Kratky
https://github.com/chkr1011/MQTTnet/blob/master/LICENSE
@@ -10,55 +10,73 @@
https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png
false
MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).
- * [Client] Received messages are now processed in the worker thread by default. Added a new setting for switching back to dedicated threads.
+ ** MQTTnet is now available at Open Collective for donations (https://opencollective.com/mqttnet). **
+* [Core] Performance optimizations.
+* [Core] Due to performance reasons the timestamp of log messages is now in UTC format.
+* [Core] Added several packet validations.
+* [Core] Log messages now contain the complete source path including parent components.
+* [Core] The adapter now has an _Endpoint_ definition as string containing remote IP and port.
+* [Client] Received messages are now processed completely in the worker thread without creating new Tasks.
+* [Client] Fixed wrong calculation for sending keep alive packets (thanks to @cstichlberger)
+* [Client] A clean disconnect (via DisconnectAsync) will no longer throw an exception.
+* [Client] Added new overloads for quick message publishing.
+* [ManagedClient] The managed client is moved to a separate nuget package.
+* [ManagedClient] Added an own message format with extended properties like ID (BREAKING CHANGE).
+* [ManagedClient] Fixed a loading issue of stored application messages (thanks to @JTrotta).
+* [ManagedClient] Added a new event which is fired when a synchronization of the subscriptions has failed.
+* [ManagedClient] Added a new event which is fired when a connection attempt has failed.
+* [ManagedClient] Exposed a new property which provides the count of not published messages (pending messages count).
* [Server] Added support for other WebSocket sub protocol formats like mqttv-3.1.1 (thanks to @israellot).
* [Server] The takeover of an existing client sessions is now treated as a _clean_ disconnect of the previous client.
+* [Server] The pending messages queue per client is now limited to 250 messages. Overflow strategy and count can be changed via options (thanks to @VladimirAkopyan)
+* [Server] Keep alive checking is now suspended while large packages are being received (and thus the client is connected). Keep alive checking continues after a large packet is received completely.
+* [Server] Rewritten the _ConnectedClients_ API and added new features for disconnecting and Endpoint information (IP etc.).
+* [Server] Added settings for disabling persistent sessions and defining a max pending messages queue size per session.
+* [Server] Persistent sessions are disabled by default (BREAKING CHANGE!).
+* [Server] Added a new interceptor which is invoked before a new message is added to the client queue.
+* [Server] Added support for Linux servers by dividing IPv4 and IPv6 support and adding new options (BREAKING CHANGE!).
+* [Server] Gracefully closed connections are no longer reported as warnings.
+* [Server] Added new overloads for initializing the ASP.NET Core integration.
Copyright Christian Kratky 2016-2018
MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin
-
-
-
+
-
-
+
-
-
-
-
-
+
-
+
-
+
-
-
-
+
+
+
+
\ No newline at end of file
diff --git a/Build/build.ps1 b/Build/build.ps1
index 021ca1d6..63bae548 100644
--- a/Build/build.ps1
+++ b/Build/build.ps1
@@ -3,6 +3,10 @@ param([string]$assemblyVersion, [string]$nugetVersion)
if ([string]::IsNullOrEmpty($assemblyVersion)) {$assemblyVersion = "0.0.1"}
if ([string]::IsNullOrEmpty($nugetVersion)) {$nugetVersion = "0.0.1"}
+Write-Host "Assembly version = $assemblyVersion"
+Write-Host "Nuget version = $nugetVersion"
+Write-Host
+
$vswhere = ${Env:\ProgramFiles(x86)} + '\Microsoft Visual Studio\Installer\vswhere'
$path = &$vswhere -latest -products * -requires Microsoft.Component.MSBuild -property installationPath
@@ -10,26 +14,34 @@ 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 /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"
+ &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="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 /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
# 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
+ &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
- Remove-Item .\NuGet -Force -Recurse -ErrorAction SilentlyContinue
+ # Build the Managed Client extension
+ &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+ &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+
+ 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
- .\NuGet.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+ .\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
+ .\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
}
\ No newline at end of file
diff --git a/Build/upload.ps1 b/Build/upload.ps1
new file mode 100644
index 00000000..a7cb1728
--- /dev/null
+++ b/Build/upload.ps1
@@ -0,0 +1,9 @@
+param([string]$apiKey)
+
+$files = Get-ChildItem -Path ".\NuGet" -Filter "*.nupkg"
+foreach ($file in $files)
+{
+ Write-Host "Uploading: " $file
+
+ .\nuget.exe push $file.Fullname $apiKey -NoSymbols -Source https://api.nuget.org/v3/index.json
+}
\ No newline at end of file
diff --git a/Extensions/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Extensions/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
deleted file mode 100644
index 938fcb4c..00000000
--- a/Extensions/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
+++ /dev/null
@@ -1,101 +0,0 @@
-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> _waitingCalls = new ConcurrentDictionary>();
- 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 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();
- 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 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();
- }
- }
-}
diff --git a/Extensions/MQTTnet.Extensions.Rpc/SampleCCode.c b/Extensions/MQTTnet.Extensions.Rpc/SampleCCode.c
deleted file mode 100644
index 5f282702..00000000
--- a/Extensions/MQTTnet.Extensions.Rpc/SampleCCode.c
+++ /dev/null
@@ -1 +0,0 @@
-
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
deleted file mode 100644
index 87b2038b..00000000
--- a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
+++ /dev/null
@@ -1,52 +0,0 @@
-using System;
-using Microsoft.AspNetCore.Builder;
-using Microsoft.Extensions.DependencyInjection;
-using System.Linq;
-using MQTTnet.Server;
-
-namespace MQTTnet.AspNetCore
-{
- public static class ApplicationBuilderExtensions
- {
- public static IApplicationBuilder UseMqttEndpoint(this IApplicationBuilder app, string path = "/mqtt")
- {
- app.UseWebSockets();
- app.Use(async (context, next) =>
- {
- if (context.Request.Path == path && context.WebSockets.IsWebSocketRequest)
- {
- string subprotocol = null;
-
- if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues))
- {
- // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
- subprotocol = requestedSubProtocolValues
- .OrderByDescending(p => p.Length)
- .FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
- }
-
- var adapter = app.ApplicationServices.GetRequiredService();
- using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subprotocol))
- {
- await adapter.AcceptWebSocketAsync(webSocket);
- }
- }
- else
- {
- await next();
- }
- });
-
- return app;
- }
-
- public static IApplicationBuilder UseMqttServer(this IApplicationBuilder app, Action configure)
- {
- var server = app.ApplicationServices.GetRequiredService();
-
- configure(server);
-
- return app;
- }
- }
-}
diff --git a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerChannel.cs b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerChannel.cs
deleted file mode 100644
index 7325a5f8..00000000
--- a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerChannel.cs
+++ /dev/null
@@ -1,60 +0,0 @@
-using System;
-using System.IO;
-using System.Net.WebSockets;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Channel;
-using MQTTnet.Implementations;
-
-namespace MQTTnet.AspNetCore
-{
- public class MqttWebSocketServerChannel : IMqttChannel, IDisposable
- {
- private WebSocket _webSocket;
-
- public MqttWebSocketServerChannel(WebSocket webSocket)
- {
- _webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket));
-
- SendStream = new WebSocketStream(_webSocket);
- ReceiveStream = SendStream;
- }
-
- public Stream SendStream { get; private set; }
- public Stream ReceiveStream { get; private set; }
-
- public Task ConnectAsync()
- {
- return Task.CompletedTask;
- }
-
- public async Task DisconnectAsync()
- {
- if (_webSocket == null)
- {
- return;
- }
-
- try
- {
- await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
- }
- finally
- {
- Dispose();
- }
- }
-
- public void Dispose()
- {
- SendStream?.Dispose();
- ReceiveStream?.Dispose();
-
- _webSocket?.Dispose();
-
- SendStream = null;
- ReceiveStream = null;
- _webSocket = null;
- }
- }
-}
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
deleted file mode 100644
index 61354cd9..00000000
--- a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
+++ /dev/null
@@ -1,31 +0,0 @@
-using System;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
-using MQTTnet.Server;
-using MQTTnet.Implementations;
-
-namespace MQTTnet.AspNetCore
-{
- public static class ServiceCollectionExtensions
- {
- public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, IMqttServerOptions options)
- {
- if (options == null) throw new ArgumentNullException(nameof(options));
-
- services.AddSingleton(options);
- services.AddSingleton(new MqttNetLogger());
- services.AddSingleton();
- services.AddSingleton(s => s.GetService());
- services.AddSingleton(s => s.GetService());
-
- services.AddSingleton();
- services.AddSingleton();
- services.AddSingleton(s => s.GetService());
- services.AddSingleton(s => s.GetService());
-
- return services;
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs
deleted file mode 100644
index 7a5c02a2..00000000
--- a/Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs
+++ /dev/null
@@ -1,22 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Packets;
-using MQTTnet.Serializer;
-
-namespace MQTTnet.Adapter
-{
- public interface IMqttChannelAdapter : IDisposable
- {
- IMqttPacketSerializer PacketSerializer { get; }
-
- Task ConnectAsync(TimeSpan timeout);
-
- Task DisconnectAsync(TimeSpan timeout);
-
- Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable packets);
-
- Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken);
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs
deleted file mode 100644
index c2f9ba43..00000000
--- a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs
+++ /dev/null
@@ -1,255 +0,0 @@
-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.Channel;
-using MQTTnet.Diagnostics;
-using MQTTnet.Exceptions;
-using MQTTnet.Internal;
-using MQTTnet.Packets;
-using MQTTnet.Serializer;
-
-namespace MQTTnet.Adapter
-{
- public sealed class MqttChannelAdapter : IMqttChannelAdapter
- {
- private const uint ErrorOperationAborted = 0x800703E3;
- private const int ReadBufferSize = 4096; // TODO: Move buffer size to config
-
- private bool _isDisposed;
- private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
- private readonly IMqttNetLogger _logger;
- private readonly IMqttChannel _channel;
-
- public MqttChannelAdapter(IMqttChannel channel, IMqttPacketSerializer serializer, IMqttNetLogger logger)
- {
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
- _channel = channel ?? throw new ArgumentNullException(nameof(channel));
- PacketSerializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
- }
-
- public IMqttPacketSerializer PacketSerializer { get; }
-
- public Task ConnectAsync(TimeSpan timeout)
- {
- ThrowIfDisposed();
- _logger.Verbose("Connecting [Timeout={0}]", timeout);
-
- return ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
- }
-
- public Task DisconnectAsync(TimeSpan timeout)
- {
- ThrowIfDisposed();
- _logger.Verbose("Disconnecting [Timeout={0}]", timeout);
-
- return ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
- }
-
- public Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable packets)
- {
- ThrowIfDisposed();
-
- return ExecuteAndWrapExceptionAsync(async () =>
- {
- await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
- try
- {
- foreach (var packet in packets)
- {
- if (cancellationToken.IsCancellationRequested)
- {
- return;
- }
-
- if (packet == null)
- {
- continue;
- }
-
- _logger.Verbose("TX >>> {0} [Timeout={1}]", packet, timeout);
-
- 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);
- }
- else
- {
- await _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false);
- }
- }
- finally
- {
- _semaphore.Release();
- }
- });
- }
-
- public async Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
- {
- ThrowIfDisposed();
-
- MqttBasePacket packet = null;
- await ExecuteAndWrapExceptionAsync(async () =>
- {
- ReceivedMqttPacket receivedMqttPacket = null;
- try
- {
- 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.Header, receivedMqttPacket.Body);
- if (packet == null)
- {
- throw new MqttProtocolViolationException("Received malformed packet.");
- }
-
- _logger.Verbose("RX <<< {0}", packet);
- }
- finally
- {
- receivedMqttPacket?.Dispose();
- }
- }).ConfigureAwait(false);
-
- return packet;
- }
-
- private static async Task ReceiveAsync(Stream stream, CancellationToken cancellationToken)
- {
- var header = await MqttPacketReader.ReadHeaderAsync(stream, cancellationToken).ConfigureAwait(false);
- if (header == null)
- {
- return null;
- }
-
- if (header.BodyLength == 0)
- {
- return new ReceivedMqttPacket(header, new MemoryStream(new byte[0], false));
- }
-
- var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream();
-
- var buffer = new byte[ReadBufferSize];
- while (body.Length < 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);
- }
-
- private static async Task ExecuteAndWrapExceptionAsync(Func action)
- {
- try
- {
- await action().ConfigureAwait(false);
- }
- catch (TaskCanceledException)
- {
- throw;
- }
- catch (OperationCanceledException)
- {
- throw;
- }
- catch (MqttCommunicationTimedOutException)
- {
- throw;
- }
- catch (MqttCommunicationException)
- {
- throw;
- }
- catch (COMException comException)
- {
- 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);
- }
- }
-
- public void Dispose()
- {
- _isDisposed = true;
- _semaphore?.Dispose();
- _channel?.Dispose();
- }
-
- private void ThrowIfDisposed()
- {
- if (_isDisposed)
- {
- throw new ObjectDisposedException(nameof(MqttChannelAdapter));
- }
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/ReceivedMqttPacket.cs b/Frameworks/MQTTnet.NetStandard/Adapter/ReceivedMqttPacket.cs
deleted file mode 100644
index a44fb54e..00000000
--- a/Frameworks/MQTTnet.NetStandard/Adapter/ReceivedMqttPacket.cs
+++ /dev/null
@@ -1,24 +0,0 @@
-using System;
-using System.IO;
-using MQTTnet.Packets;
-
-namespace MQTTnet.Adapter
-{
- public sealed class ReceivedMqttPacket : IDisposable
- {
- public ReceivedMqttPacket(MqttPacketHeader header, MemoryStream body)
- {
- Header = header ?? throw new ArgumentNullException(nameof(header));
- Body = body ?? throw new ArgumentNullException(nameof(body));
- }
-
- public MqttPacketHeader Header { get; }
-
- public MemoryStream Body { get; }
-
- public void Dispose()
- {
- Body?.Dispose();
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs b/Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs
deleted file mode 100644
index b1bf62f5..00000000
--- a/Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs
+++ /dev/null
@@ -1,41 +0,0 @@
-using System;
-using System.Threading.Tasks;
-using MQTTnet.Protocol;
-
-namespace MQTTnet
-{
- public static class ApplicationMessagePublisherExtensions
- {
- public static Task PublishAsync(this IApplicationMessagePublisher publisher, params MqttApplicationMessage[] applicationMessages)
- {
- if (publisher == null) throw new ArgumentNullException(nameof(publisher));
- if (applicationMessages == null) throw new ArgumentNullException(nameof(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());
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs b/Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs
deleted file mode 100644
index b332884d..00000000
--- a/Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs
+++ /dev/null
@@ -1,15 +0,0 @@
-using System;
-using System.IO;
-using System.Threading.Tasks;
-
-namespace MQTTnet.Channel
-{
- public interface IMqttChannel : IDisposable
- {
- Stream SendStream { get; }
- Stream ReceiveStream { get; }
-
- Task ConnectAsync();
- Task DisconnectAsync();
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs
deleted file mode 100644
index 8b09fbce..00000000
--- a/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs
+++ /dev/null
@@ -1,97 +0,0 @@
-using System;
-using System.Collections.Concurrent;
-using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
-using MQTTnet.Exceptions;
-using MQTTnet.Internal;
-using MQTTnet.Packets;
-
-namespace MQTTnet.Client
-{
- public class MqttPacketDispatcher
- {
- private readonly ConcurrentDictionary>> _awaiters = new ConcurrentDictionary>>();
- private readonly IMqttNetLogger _logger;
-
- public MqttPacketDispatcher(IMqttNetLogger logger)
- {
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
- }
-
- public async Task WaitForPacketAsync(Type responseType, ushort? identifier, TimeSpan timeout)
- {
- var packetAwaiter = AddPacketAwaiter(responseType, identifier);
- try
- {
- return await packetAwaiter.Task.TimeoutAfter(timeout).ConfigureAwait(false);
- }
- catch (MqttCommunicationTimedOutException)
- {
- _logger.Warning("Timeout while waiting for packet of type '{0}'.", responseType.Name);
- throw;
- }
- finally
- {
- RemovePacketAwaiter(responseType, identifier);
- }
- }
-
- public void Dispatch(MqttBasePacket packet)
- {
- if (packet == null) throw new ArgumentNullException(nameof(packet));
-
- var type = packet.GetType();
-
- if (_awaiters.TryGetValue(type, out var byId))
- {
- ushort? identifier = 0;
- if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
- {
- identifier = packetWithIdentifier.PacketIdentifier;
- }
-
- if (byId.TryRemove(identifier.Value, out var tcs))
- {
- tcs.TrySetResult(packet);
- return;
- }
- }
-
- throw new InvalidOperationException($"Packet of type '{type.Name}' not handled or dispatched.");
- }
-
- public void Reset()
- {
- _awaiters.Clear();
- }
-
- private TaskCompletionSource AddPacketAwaiter(Type responseType, ushort? identifier)
- {
- var tcs = new TaskCompletionSource();
-
- if (!identifier.HasValue)
- {
- identifier = 0;
- }
-
- var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary>());
- if (!byId.TryAdd(identifier.Value, tcs))
- {
- throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{responseType}' with identifier {identifier}.");
- }
-
- return tcs;
- }
-
- private void RemovePacketAwaiter(Type responseType, ushort? identifier)
- {
- if (!identifier.HasValue)
- {
- identifier = 0;
- }
-
- var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary>());
- byId.TryRemove(identifier.Value, out var _);
- }
- }
-}
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs
deleted file mode 100644
index 651abe5b..00000000
--- a/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs
+++ /dev/null
@@ -1,8 +0,0 @@
-namespace MQTTnet.Client
-{
- public enum MqttReceivedApplicationMessageProcessingMode
- {
- SingleThread,
- DedicatedThread
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Diagnostics/IMqttNetLogger.cs b/Frameworks/MQTTnet.NetStandard/Diagnostics/IMqttNetLogger.cs
deleted file mode 100644
index 6ed16de2..00000000
--- a/Frameworks/MQTTnet.NetStandard/Diagnostics/IMqttNetLogger.cs
+++ /dev/null
@@ -1,21 +0,0 @@
-using System;
-
-namespace MQTTnet.Diagnostics
-{
- public interface IMqttNetLogger
- {
- event EventHandler LogMessagePublished;
-
- void Verbose(string message, params object[] parameters);
-
- void Info(string message, params object[] parameters);
-
- void Warning(Exception exception, string message, params object[] parameters);
-
- void Warning(string message, params object[] parameters);
-
- void Error(Exception exception, string message, params object[] parameters);
-
- void Error(string message, params object[] parameters);
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs b/Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs
deleted file mode 100644
index 53598b2b..00000000
--- a/Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs
+++ /dev/null
@@ -1,74 +0,0 @@
-using System;
-
-namespace MQTTnet.Diagnostics
-{
- public class MqttNetLogger : IMqttNetLogger
- {
- private readonly string _logId;
-
- public MqttNetLogger(string logId = null)
- {
- _logId = logId;
- }
-
- public event EventHandler LogMessagePublished;
-
- public void Verbose(string message, params object[] parameters)
- {
- Publish(MqttNetLogLevel.Verbose, null, message, parameters);
- }
-
- public void Info(string message, params object[] parameters)
- {
- Publish(MqttNetLogLevel.Info, null, message, parameters);
- }
-
- public void Warning(Exception exception, string message, params object[] parameters)
- {
- Publish(MqttNetLogLevel.Warning, exception, message, parameters);
- }
-
- public void Warning(string message, params object[] parameters)
- {
- Warning(null, message, parameters);
- }
-
- public void Error(Exception exception, string message, params object[] parameters)
- {
- Publish(MqttNetLogLevel.Error, exception, message, parameters);
- }
-
- public void Error(string message, params object[] parameters)
- {
- Warning(null, message, parameters);
- }
-
- private void Publish(MqttNetLogLevel logLevel, Exception exception, string message, object[] parameters)
- {
- var hasLocalListeners = LogMessagePublished != null;
- var hasGlobalListeners = MqttNetGlobalLogger.HasListeners;
-
- if (!hasLocalListeners && !hasGlobalListeners)
- {
- return;
- }
-
- if (parameters.Length > 0)
- {
- message = string.Format(message, parameters);
- }
-
- var traceMessage = new MqttNetLogMessage(_logId, DateTime.Now, Environment.CurrentManagedThreadId, typeof(TSource).Name, logLevel, message, exception);
-
- if (hasGlobalListeners)
- {
- MqttNetGlobalLogger.Publish(traceMessage);
- }
-
- if (hasLocalListeners)
- {
- LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(traceMessage));
- }
- }
- }
-}
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.NetStandard/Exceptions/MqttCommunicationTimedOutException.cs b/Frameworks/MQTTnet.NetStandard/Exceptions/MqttCommunicationTimedOutException.cs
deleted file mode 100644
index 7d0adcd4..00000000
--- a/Frameworks/MQTTnet.NetStandard/Exceptions/MqttCommunicationTimedOutException.cs
+++ /dev/null
@@ -1,6 +0,0 @@
-namespace MQTTnet.Exceptions
-{
- public sealed class MqttCommunicationTimedOutException : MqttCommunicationException
- {
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/IApplicationMessagePublisher.cs b/Frameworks/MQTTnet.NetStandard/IApplicationMessagePublisher.cs
deleted file mode 100644
index e137669e..00000000
--- a/Frameworks/MQTTnet.NetStandard/IApplicationMessagePublisher.cs
+++ /dev/null
@@ -1,10 +0,0 @@
-using System.Collections.Generic;
-using System.Threading.Tasks;
-
-namespace MQTTnet
-{
- public interface IApplicationMessagePublisher
- {
- Task PublishAsync(IEnumerable applicationMessages);
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.Uwp.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.Uwp.cs
deleted file mode 100644
index e71d6fff..00000000
--- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.Uwp.cs
+++ /dev/null
@@ -1,75 +0,0 @@
-#if WINDOWS_UWP
-using System;
-using System.Threading.Tasks;
-using Windows.Networking.Sockets;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
-using MQTTnet.Serializer;
-using MQTTnet.Server;
-
-namespace MQTTnet.Implementations
-{
- public class MqttTcpServerAdapter : IMqttServerAdapter, IDisposable
- {
- private readonly IMqttNetLogger _logger;
- private StreamSocketListener _defaultEndpointSocket;
-
- public MqttTcpServerAdapter(IMqttNetLogger logger)
- {
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
- }
-
- public event EventHandler ClientAccepted;
-
- public async Task StartAsync(IMqttServerOptions options)
- {
- if (options == null) throw new ArgumentNullException(nameof(options));
-
- if (_defaultEndpointSocket != null) throw new InvalidOperationException("Server is already started.");
-
- if (options.DefaultEndpointOptions.IsEnabled)
- {
- _defaultEndpointSocket = new StreamSocketListener();
- await _defaultEndpointSocket.BindServiceNameAsync(options.GetDefaultEndpointPort().ToString(), SocketProtectionLevel.PlainSocket);
- _defaultEndpointSocket.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync;
- }
-
- if (options.TlsEndpointOptions.IsEnabled)
- {
- throw new NotSupportedException("TLS servers are not supported for UWP apps.");
- }
- }
-
- public Task StopAsync()
- {
- if (_defaultEndpointSocket != null)
- {
- _defaultEndpointSocket.ConnectionReceived -= AcceptDefaultEndpointConnectionsAsync;
- }
-
- _defaultEndpointSocket?.Dispose();
- _defaultEndpointSocket = null;
-
- return Task.FromResult(0);
- }
-
- public void Dispose()
- {
- StopAsync();
- }
-
- private void AcceptDefaultEndpointConnectionsAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
- {
- try
- {
- var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket), new MqttPacketSerializer(), _logger);
- ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
- }
- catch (Exception exception)
- {
- _logger.Error(exception, "Error while accepting connection at default endpoint.");
- }
- }
- }
-}
-#endif
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.cs
deleted file mode 100644
index dc33a3f2..00000000
--- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.cs
+++ /dev/null
@@ -1,162 +0,0 @@
-#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
-using System;
-using System.Net;
-using System.Net.Security;
-using System.Net.Sockets;
-using System.Security.Authentication;
-using System.Security.Cryptography.X509Certificates;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
-using MQTTnet.Serializer;
-using MQTTnet.Server;
-
-namespace MQTTnet.Implementations
-{
- public class MqttTcpServerAdapter : IMqttServerAdapter, IDisposable
- {
- private readonly IMqttNetLogger _logger;
-
- private CancellationTokenSource _cancellationTokenSource;
- private Socket _defaultEndpointSocket;
- private Socket _tlsEndpointSocket;
- private X509Certificate2 _tlsCertificate;
-
- public MqttTcpServerAdapter(IMqttNetLogger logger)
- {
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
- }
-
- public event EventHandler ClientAccepted;
-
- public Task StartAsync(IMqttServerOptions options)
- {
- if (_cancellationTokenSource != null) throw new InvalidOperationException("Server is already started.");
-
- _cancellationTokenSource = new CancellationTokenSource();
-
- if (options.DefaultEndpointOptions.IsEnabled)
- {
- _defaultEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
- _defaultEndpointSocket.Bind(new IPEndPoint(options.DefaultEndpointOptions.BoundIPAddress, options.GetDefaultEndpointPort()));
- _defaultEndpointSocket.Listen(options.ConnectionBacklog);
-
- Task.Run(async () => await AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false);
- }
-
- if (options.TlsEndpointOptions.IsEnabled)
- {
- if (options.TlsEndpointOptions.Certificate == null)
- {
- throw new ArgumentException("TLS certificate is not set.");
- }
-
- _tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate);
- if (!_tlsCertificate.HasPrivateKey)
- {
- throw new InvalidOperationException("The certificate for TLS encryption must contain the private key.");
- }
-
- _tlsEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
- _tlsEndpointSocket.Bind(new IPEndPoint(options.TlsEndpointOptions.BoundIPAddress, options.GetTlsEndpointPort()));
- _tlsEndpointSocket.Listen(options.ConnectionBacklog);
-
- Task.Run(async () => await AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false);
- }
-
- return Task.FromResult(0);
- }
-
- public Task StopAsync()
- {
- _cancellationTokenSource?.Cancel(false);
- _cancellationTokenSource?.Dispose();
- _cancellationTokenSource = null;
-
- _defaultEndpointSocket?.Dispose();
- _defaultEndpointSocket = null;
-
- _tlsCertificate = null;
-
- _tlsEndpointSocket?.Dispose();
- _tlsEndpointSocket = null;
-
- return Task.FromResult(0);
- }
-
- public void Dispose()
- {
- StopAsync();
- }
-
- private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cancellationToken)
- {
- while (!cancellationToken.IsCancellationRequested)
- {
- try
- {
- //todo: else branch can be used with min dependency NET46
-#if NET452 || NET461
- var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false);
-#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 (Exception exception)
- {
- if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
- {
- return;
- }
-
- _logger.Error(exception, "Error while accepting connection at default endpoint.");
- await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
- }
- }
- }
-
- private async Task AcceptTlsEndpointConnectionsAsync(CancellationToken cancellationToken)
- {
- while (!cancellationToken.IsCancellationRequested)
- {
- try
- {
-#if NET452 || NET461
- var clientSocket = await Task.Factory.FromAsync(_tlsEndpointSocket.BeginAccept, _tlsEndpointSocket.EndAccept, null).ConfigureAwait(false);
-#else
- var clientSocket = await _tlsEndpointSocket.AcceptAsync().ConfigureAwait(false);
-#endif
-
- 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 (Exception exception)
- {
- if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
- {
- return;
- }
-
- _logger.Error(exception, "Error while accepting connection at TLS endpoint.");
- await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
- }
- }
- }
- }
-}
-#endif
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs b/Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs
deleted file mode 100644
index bad03935..00000000
--- a/Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs
+++ /dev/null
@@ -1,137 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Linq;
-using System.Net.WebSockets;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Exceptions;
-
-namespace MQTTnet.Implementations
-{
- public class WebSocketStream : Stream
- {
- private readonly byte[] _chunckBuffer = new byte[MqttWebSocketChannel.BufferSize];
- private readonly Queue _buffer = new Queue(MqttWebSocketChannel.BufferSize);
- private readonly WebSocket _webSocket;
-
- public WebSocketStream(WebSocket webSocket)
- {
- _webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket));
- }
-
- public override bool CanRead => true;
-
- public override bool CanSeek => false;
-
- public override bool CanWrite => true;
-
- public override long Length => throw new NotSupportedException();
-
- public override long Position
- {
- get => throw new NotSupportedException();
- set => throw new NotSupportedException();
- }
-
- public override void Flush()
- {
- }
-
- public override Task FlushAsync(CancellationToken cancellationToken)
- {
- return Task.FromResult(0);
- }
-
- public override int Read(byte[] buffer, int offset, int count)
- {
- return ReadAsync(buffer, offset, count).GetAwaiter().GetResult();
- }
-
- public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- var bytesRead = 0;
-
- // Use existing date from buffer.
- while (count > 0 && _buffer.Any())
- {
- 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).ConfigureAwait(false);
-
- while (count > 0 && _buffer.Any())
- {
- buffer[offset] = _buffer.Dequeue();
- count--;
- bytesRead++;
- offset++;
- }
-
- if (count == 0)
- {
- return bytesRead;
- }
- }
-
- if (_webSocket.State == WebSocketState.Closed)
- {
- throw new MqttCommunicationException("WebSocket connection closed.");
- }
-
- return bytesRead;
- }
-
- public override void Write(byte[] buffer, int offset, int count)
- {
- WriteAsync(buffer, offset, count).GetAwaiter().GetResult();
- }
-
- public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- return _webSocket.SendAsync(new ArraySegment(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken);
- }
-
- public override long Seek(long offset, SeekOrigin origin)
- {
- throw new NotSupportedException();
- }
-
- public override void SetLength(long value)
- {
- throw new NotSupportedException();
- }
-
- private async Task FetchChunkAsync(CancellationToken cancellationToken)
- {
- var response = await _webSocket.ReceiveAsync(new ArraySegment(_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.");
- }
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Internal/TaskExtensions.cs b/Frameworks/MQTTnet.NetStandard/Internal/TaskExtensions.cs
deleted file mode 100644
index 876c5495..00000000
--- a/Frameworks/MQTTnet.NetStandard/Internal/TaskExtensions.cs
+++ /dev/null
@@ -1,78 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Exceptions;
-
-namespace MQTTnet.Internal
-{
- public static class TaskExtensions
- {
- public static async Task TimeoutAfter(this Task task, TimeSpan timeout)
- {
- if (task == null) throw new ArgumentNullException(nameof(task));
-
- using (var timeoutCts = new CancellationTokenSource())
- {
- try
- {
- var timeoutTask = Task.Delay(timeout, timeoutCts.Token);
- var finishedTask = await Task.WhenAny(timeoutTask, task).ConfigureAwait(false);
-
- if (finishedTask == timeoutTask)
- {
- throw new MqttCommunicationTimedOutException();
- }
-
- if (task.IsCanceled)
- {
- throw new TaskCanceledException();
- }
-
- if (task.IsFaulted)
- {
- throw new MqttCommunicationException(task.Exception?.GetBaseException());
- }
- }
- finally
- {
- timeoutCts.Cancel();
- }
- }
- }
-
- public static async Task TimeoutAfter(this Task task, TimeSpan timeout)
- {
- if (task == null) throw new ArgumentNullException(nameof(task));
-
- using (var timeoutCts = new CancellationTokenSource())
- {
- try
- {
- var timeoutTask = Task.Delay(timeout, timeoutCts.Token);
- var finishedTask = await Task.WhenAny(timeoutTask, task).ConfigureAwait(false);
-
- if (finishedTask == timeoutTask)
- {
- throw new MqttCommunicationTimedOutException();
- }
-
- if (task.IsCanceled)
- {
- throw new TaskCanceledException();
- }
-
- if (task.IsFaulted)
- {
- throw new MqttCommunicationException(task.Exception.GetBaseException());
- }
-
- return task.Result;
- }
- finally
- {
- timeoutCts.Cancel();
- }
- }
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/ManagedClient/IManagedMqttClientStorage.cs b/Frameworks/MQTTnet.NetStandard/ManagedClient/IManagedMqttClientStorage.cs
deleted file mode 100644
index 28bf5703..00000000
--- a/Frameworks/MQTTnet.NetStandard/ManagedClient/IManagedMqttClientStorage.cs
+++ /dev/null
@@ -1,12 +0,0 @@
-using System.Collections.Generic;
-using System.Threading.Tasks;
-
-namespace MQTTnet.ManagedClient
-{
- public interface IManagedMqttClientStorage
- {
- Task SaveQueuedMessagesAsync(IList messages);
-
- Task> LoadQueuedMessagesAsync();
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientStorageManager.cs b/Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientStorageManager.cs
deleted file mode 100644
index 0e71776e..00000000
--- a/Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientStorageManager.cs
+++ /dev/null
@@ -1,67 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace MQTTnet.ManagedClient
-{
- public class ManagedMqttClientStorageManager
- {
- private readonly List _applicationMessages = new List();
- private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
- private readonly IManagedMqttClientStorage _storage;
-
- public ManagedMqttClientStorageManager(IManagedMqttClientStorage storage)
- {
- _storage = storage ?? throw new ArgumentNullException(nameof(storage));
- }
-
- public async Task LoadQueuedMessagesAsync()
- {
- var loadedMessages = await _storage.LoadQueuedMessagesAsync().ConfigureAwait(false);
- foreach (var loadedMessage in loadedMessages)
- {
- _applicationMessages.Add(loadedMessage);
- }
- }
-
- public async Task AddAsync(MqttApplicationMessage applicationMessage)
- {
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- _applicationMessages.Add(applicationMessage);
- await SaveAsync().ConfigureAwait(false);
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- public async Task RemoveAsync(MqttApplicationMessage applicationMessage)
- {
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- var index = _applicationMessages.IndexOf(applicationMessage);
- if (index == -1)
- {
- return;
- }
-
- _applicationMessages.RemoveAt(index);
- await SaveAsync().ConfigureAwait(false);
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- private Task SaveAsync()
- {
- return _storage.SaveQueuedMessagesAsync(_applicationMessages);
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/MqttApplicationMessage.cs b/Frameworks/MQTTnet.NetStandard/MqttApplicationMessage.cs
deleted file mode 100644
index 387acb21..00000000
--- a/Frameworks/MQTTnet.NetStandard/MqttApplicationMessage.cs
+++ /dev/null
@@ -1,29 +0,0 @@
-using System;
-using MQTTnet.Protocol;
-
-namespace MQTTnet
-{
- public sealed class MqttApplicationMessage
- {
- public MqttApplicationMessage()
- {
- }
-
- [Obsolete("Use object initializer or _MqttApplicationMessageBuilder_ instead.")]
- public MqttApplicationMessage(string topic, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, bool retain)
- {
- Topic = topic ?? throw new ArgumentNullException(nameof(topic));
- Payload = payload ?? throw new ArgumentNullException(nameof(payload));
- QualityOfServiceLevel = qualityOfServiceLevel;
- Retain = retain;
- }
-
- public string Topic { get; set; }
-
- public byte[] Payload { get; set; }
-
- public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
-
- public bool Retain { get; set; }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttPacketHeader.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttPacketHeader.cs
deleted file mode 100644
index 01c26ecd..00000000
--- a/Frameworks/MQTTnet.NetStandard/Packets/MqttPacketHeader.cs
+++ /dev/null
@@ -1,13 +0,0 @@
-using MQTTnet.Protocol;
-
-namespace MQTTnet.Packets
-{
- public class MqttPacketHeader
- {
- public MqttControlPacketType ControlPacketType { get; set; }
-
- public byte FixedHeader { get; set; }
-
- public int BodyLength { get; set; }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttPubAckPacket.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttPubAckPacket.cs
deleted file mode 100644
index e7fe57b7..00000000
--- a/Frameworks/MQTTnet.NetStandard/Packets/MqttPubAckPacket.cs
+++ /dev/null
@@ -1,10 +0,0 @@
-namespace MQTTnet.Packets
-{
- public sealed class MqttPubAckPacket : MqttBasePublishPacket
- {
- public override string ToString()
- {
- return "PubAck";
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/ByteReader.cs b/Frameworks/MQTTnet.NetStandard/Serializer/ByteReader.cs
deleted file mode 100644
index ec1e3ea2..00000000
--- a/Frameworks/MQTTnet.NetStandard/Serializer/ByteReader.cs
+++ /dev/null
@@ -1,48 +0,0 @@
-using System;
-
-namespace MQTTnet.Serializer
-{
- public sealed class ByteReader
- {
- private readonly int _source;
- private int _index;
-
- public ByteReader(int source)
- {
- _source = source;
- }
-
- public bool Read()
- {
- if (_index >= 8)
- {
- throw new InvalidOperationException("End of byte reached.");
- }
-
- var result = ((1 << _index) & _source) > 0;
- _index++;
- return result;
- }
-
- public int Read(int count)
- {
- if (_index + count > 8)
- {
- throw new InvalidOperationException("End of byte will be reached.");
- }
-
- var result = 0;
- for (var i = 0; i < count; i++)
- {
- if (((1 << _index) & _source) > 0)
- {
- result |= 1 << i;
- }
-
- _index++;
- }
-
- return result;
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/ByteWriter.cs b/Frameworks/MQTTnet.NetStandard/Serializer/ByteWriter.cs
deleted file mode 100644
index af961e1a..00000000
--- a/Frameworks/MQTTnet.NetStandard/Serializer/ByteWriter.cs
+++ /dev/null
@@ -1,36 +0,0 @@
-using System;
-
-namespace MQTTnet.Serializer
-{
- public sealed class ByteWriter
- {
- private int _index;
- private int _byte;
-
- public byte Value => (byte)_byte;
-
- public void Write(int @byte, int count)
- {
- for (var i = 0; i < count; i++)
- {
- var value = ((1 << i) & @byte) > 0;
- Write(value);
- }
- }
-
- public void Write(bool bit)
- {
- if (_index >= 8)
- {
- throw new InvalidOperationException("End of the byte reached.");
- }
-
- if (bit)
- {
- _byte |= 1 << _index;
- }
-
- _index++;
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs b/Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs
deleted file mode 100644
index 6577b0ad..00000000
--- a/Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs
+++ /dev/null
@@ -1,16 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
-using MQTTnet.Packets;
-
-namespace MQTTnet.Serializer
-{
- public interface IMqttPacketSerializer
- {
- MqttProtocolVersion ProtocolVersion { get; set; }
-
- ICollection> Serialize(MqttBasePacket mqttPacket);
-
- MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body);
- }
-}
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs
deleted file mode 100644
index bd43f7d3..00000000
--- a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs
+++ /dev/null
@@ -1,129 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Exceptions;
-using MQTTnet.Packets;
-using MQTTnet.Protocol;
-
-namespace MQTTnet.Serializer
-{
- public sealed class MqttPacketReader : BinaryReader
- {
- private readonly MqttPacketHeader _header;
-
- public MqttPacketReader(MqttPacketHeader header, Stream bodyStream)
- : base(bodyStream, Encoding.UTF8, true)
- {
- _header = header;
- }
-
- public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength;
-
- public static async Task ReadHeaderAsync(Stream stream, CancellationToken cancellationToken)
- {
- if (cancellationToken.IsCancellationRequested)
- {
- return null;
- }
-
- // 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 = await ReadBodyLengthAsync(stream, cancellationToken).ConfigureAwait(false);
-
- return new MqttPacketHeader
- {
- FixedHeader = fixedHeader,
- ControlPacketType = controlPacketType,
- BodyLength = bodyLength
- };
- }
-
- public override ushort ReadUInt16()
- {
- var buffer = ReadBytes(2);
-
- var temp = buffer[0];
- buffer[0] = buffer[1];
- buffer[1] = temp;
-
- return BitConverter.ToUInt16(buffer, 0);
- }
-
- public string ReadStringWithLengthPrefix()
- {
- var buffer = ReadWithLengthPrefix();
- if (buffer.Length == 0)
- {
- return string.Empty;
- }
-
- return Encoding.UTF8.GetString(buffer, 0, buffer.Length);
- }
-
- public byte[] ReadWithLengthPrefix()
- {
- var length = ReadUInt16();
- if (length == 0)
- {
- return new byte[0];
- }
-
- return ReadBytes(length);
- }
-
- public byte[] ReadRemainingData()
- {
- return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
- }
-
- private static async Task ReadBodyLengthAsync(Stream stream, CancellationToken cancellationToken)
- {
- // 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();
- do
- {
- if (cancellationToken.IsCancellationRequested)
- {
- throw new TaskCanceledException();
- }
-
- 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 = buffer[0];
- readBytes.Add(encodedByte);
-
- value += (byte)(encodedByte & 127) * multiplier;
- if (multiplier > 128 * 128 * 128)
- {
- throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
- }
-
- multiplier *= 128;
- } while ((encodedByte & 128) != 0);
-
- return value;
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs
deleted file mode 100644
index cb3d4589..00000000
--- a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs
+++ /dev/null
@@ -1,81 +0,0 @@
-using System;
-using System.IO;
-using System.Text;
-using MQTTnet.Protocol;
-
-namespace MQTTnet.Serializer
-{
- public sealed class MqttPacketWriter : BinaryWriter
- {
- public MqttPacketWriter(Stream stream)
- : base(stream, Encoding.UTF8, true)
- {
- }
-
- public static byte BuildFixedHeader(MqttControlPacketType packetType, byte flags = 0)
- {
- var fixedHeader = (int)packetType << 4;
- fixedHeader |= flags;
- return (byte)fixedHeader;
- }
-
- public override void Write(ushort value)
- {
- var buffer = BitConverter.GetBytes(value);
- Write(buffer[1], buffer[0]);
- }
-
- public new void Write(params byte[] values)
- {
- base.Write(values);
- }
-
- public new void Write(byte value)
- {
- base.Write(value);
- }
-
- public void Write(ByteWriter value)
- {
- if (value == null) throw new ArgumentNullException(nameof(value));
-
- Write(value.Value);
- }
-
- public void WriteWithLengthPrefix(string value)
- {
- WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty));
- }
-
- public void WriteWithLengthPrefix(byte[] value)
- {
- var length = (ushort)value.Length;
-
- Write(length);
- Write(value);
- }
-
- public static void WriteRemainingLength(int length, BinaryWriter target)
- {
- if (length == 0)
- {
- target.Write((byte)0);
- return;
- }
-
- // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
- var x = length;
- do
- {
- var encodedByte = x % 128;
- x = x / 128;
- if (x > 0)
- {
- encodedByte = encodedByte | 128;
- }
-
- target.Write((byte)encodedByte);
- } while (x > 0);
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs b/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs
deleted file mode 100644
index 4fa8e4ea..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs
+++ /dev/null
@@ -1,18 +0,0 @@
-using System;
-using MQTTnet.Serializer;
-
-namespace MQTTnet.Server
-{
- public class ConnectedMqttClient
- {
- public string ClientId { get; set; }
-
- public MqttProtocolVersion ProtocolVersion { get; set; }
-
- public TimeSpan LastPacketReceived { get; set; }
-
- public TimeSpan LastNonKeepAlivePacketReceived { get; set; }
-
- public int PendingApplicationMessages { get; set; }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientConnectedEventArgs.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientConnectedEventArgs.cs
deleted file mode 100644
index 1ca3f406..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientConnectedEventArgs.cs
+++ /dev/null
@@ -1,14 +0,0 @@
-using System;
-
-namespace MQTTnet.Server
-{
- public class MqttClientConnectedEventArgs : EventArgs
- {
- public MqttClientConnectedEventArgs(ConnectedMqttClient client)
- {
- Client = client ?? throw new ArgumentNullException(nameof(client));
- }
-
- public ConnectedMqttClient Client { get; }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs
deleted file mode 100644
index 1602a4d1..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs
+++ /dev/null
@@ -1,141 +0,0 @@
-using System;
-using System.Collections.Concurrent;
-using System.Runtime.CompilerServices;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
-using MQTTnet.Exceptions;
-using MQTTnet.Packets;
-using MQTTnet.Protocol;
-
-namespace MQTTnet.Server
-{
- public sealed class MqttClientPendingMessagesQueue : IDisposable
- {
- private readonly ConcurrentQueue _queue = new ConcurrentQueue();
- private readonly SemaphoreSlim _queueWaitSemaphore = new SemaphoreSlim(0);
- private readonly IMqttServerOptions _options;
- private readonly MqttClientSession _clientSession;
- private readonly IMqttNetLogger _logger;
-
- private Task _workerTask;
-
- public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetLogger logger)
- {
- _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));
-
- if (cancellationToken.IsCancellationRequested)
- {
- return;
- }
-
- _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.Enqueue(packet);
- _queueWaitSemaphore.Release();
-
- _logger.Verbose("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
- }
-
- private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
- {
- try
- {
- while (!cancellationToken.IsCancellationRequested)
- {
- await SendNextQueuedPacketAsync(adapter, cancellationToken).ConfigureAwait(false);
- }
- }
- catch (OperationCanceledException)
- {
- }
- catch (Exception exception)
- {
- _logger.Error(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _clientSession.ClientId);
- }
- }
-
- private async Task SendNextQueuedPacketAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
- {
- MqttBasePacket packet = null;
- try
- {
- await _queueWaitSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
- if (!_queue.TryDequeue(out packet))
- {
- throw new InvalidOperationException(); // should not happen
- }
-
- if (cancellationToken.IsCancellationRequested)
- {
- return;
- }
-
- await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false);
-
- _logger.Verbose("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
- }
- catch (Exception exception)
- {
- if (exception is MqttCommunicationTimedOutException)
- {
- _logger.Warning(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _clientSession.ClientId);
- }
- else if (exception is MqttCommunicationException)
- {
- _logger.Warning(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _clientSession.ClientId);
- }
- else if (exception is OperationCanceledException)
- {
- }
- else
- {
- _logger.Error(exception, "Sending publish packet failed (ClientId: {0}).", _clientSession.ClientId);
- }
-
- if (packet is MqttPublishPacket publishPacket)
- {
- if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
- {
- publishPacket.Dup = true;
- _queue.Enqueue(packet);
- _queueWaitSemaphore.Release();
- }
- }
-
- if (!cancellationToken.IsCancellationRequested)
- {
- await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
- }
- }
- }
-
- public void Dispose()
- {
- _queueWaitSemaphore?.Dispose();
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs
deleted file mode 100644
index f86f72a1..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs
+++ /dev/null
@@ -1,349 +0,0 @@
-using System;
-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;
-using MQTTnet.Packets;
-using MQTTnet.Protocol;
-using MQTTnet.Serializer;
-
-namespace MQTTnet.Server
-{
- public sealed class MqttClientSession : IDisposable
- {
- private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
- private readonly IMqttServerOptions _options;
- private readonly IMqttNetLogger _logger;
- private readonly MqttRetainedMessagesManager _retainedMessagesManager;
-
- private IMqttChannelAdapter _adapter;
- private CancellationTokenSource _cancellationTokenSource;
- private MqttApplicationMessage _willMessage;
- private bool _wasCleanDisconnect;
-
- public MqttClientSession(
- string clientId,
- IMqttServerOptions options,
- MqttRetainedMessagesManager retainedMessagesManager,
- IMqttNetLogger logger)
- {
- _options = options ?? throw new ArgumentNullException(nameof(options));
- _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
-
- ClientId = clientId;
-
- KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeoutAsync, _logger);
- SubscriptionsManager = new MqttClientSubscriptionsManager(_options, clientId);
- PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
- }
-
- public Func 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 bool IsConnected => _adapter != null;
-
- public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
- {
- if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket));
- if (adapter == null) throw new ArgumentNullException(nameof(adapter));
-
- try
- {
- var cancellationTokenSource = new CancellationTokenSource();
-
- _wasCleanDisconnect = false;
- _willMessage = connectPacket.WillMessage;
- _adapter = adapter;
- _cancellationTokenSource = cancellationTokenSource;
-
- PendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
- KeepAliveMonitor.Start(connectPacket.KeepAlivePeriod, cancellationTokenSource.Token);
-
- await ReceivePacketsAsync(adapter, cancellationTokenSource.Token).ConfigureAwait(false);
- }
- catch (OperationCanceledException)
- {
- }
- catch (MqttCommunicationException exception)
- {
- _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
- }
- catch (Exception exception)
- {
- _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
- }
-
- return _wasCleanDisconnect;
- }
-
- public async Task StopAsync(MqttClientDisconnectType type)
- {
- try
- {
- if (_cancellationTokenSource == null)
- {
- return;
- }
-
- _wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
-
- _cancellationTokenSource?.Cancel(false);
-
- PendingMessagesQueue.WaitForCompletion();
- KeepAliveMonitor.WaitForCompletion();
-
- _cancellationTokenSource?.Dispose();
- _cancellationTokenSource = null;
-
- _adapter = null;
-
- _logger.Info("Client '{0}': Session stopped.", ClientId);
- }
- finally
- {
- var willMessage = _willMessage;
- _willMessage = null; // clear willmessage so it is send just once
-
- if (willMessage != null && !_wasCleanDisconnect)
- {
- await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
- }
- }
- }
-
- public async Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage)
- {
- if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-
- var result = await SubscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
- if (!result.IsSubscribed)
- {
- return;
- }
-
- var publishPacket = applicationMessage.ToPublishPacket();
- publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
-
- if (publishPacket.QualityOfServiceLevel > 0)
- {
- publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
- }
-
- PendingMessagesQueue.Enqueue(publishPacket);
- }
-
- public async Task SubscribeAsync(IList 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 topicFilters)
- {
- if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
-
- return SubscriptionsManager.UnsubscribeAsync(new MqttUnsubscribePacket
- {
- TopicFilters = topicFilters
- });
- }
-
- public void Dispose()
- {
- ApplicationMessageReceivedCallback = null;
-
- SubscriptionsManager?.Dispose();
- PendingMessagesQueue?.Dispose();
-
- _cancellationTokenSource?.Dispose();
- }
-
- private Task StopDueToKeepAliveTimeoutAsync()
- {
- _logger.Info("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
- return StopAsync(MqttClientDisconnectType.NotClean);
- }
-
- private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
- {
- try
- {
- while (!cancellationToken.IsCancellationRequested)
- {
- var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
- KeepAliveMonitor.PacketReceived(packet);
- await ProcessReceivedPacketAsync(adapter, packet, cancellationToken).ConfigureAwait(false);
- }
- }
- catch (OperationCanceledException)
- {
- }
- catch (Exception exception)
- {
- if (exception is MqttCommunicationException)
- {
- _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
- }
- else
- {
- _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
- }
-
- await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
- }
- }
-
- private Task ProcessReceivedPacketAsync(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
- {
- if (packet is MqttPublishPacket publishPacket)
- {
- return HandleIncomingPublishPacketAsync(adapter, publishPacket, cancellationToken);
- }
-
- if (packet is MqttPingReqPacket)
- {
- return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttPingRespPacket() });
- }
-
- if (packet is MqttPubRelPacket pubRelPacket)
- {
- return HandleIncomingPubRelPacketAsync(adapter, pubRelPacket, cancellationToken);
- }
-
- if (packet is MqttPubRecPacket pubRecPacket)
- {
- var responsePacket = new MqttPubRelPacket
- {
- PacketIdentifier = pubRecPacket.PacketIdentifier
- };
-
- return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { responsePacket });
- }
-
- if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
- {
- // Discard message.
- return Task.FromResult(0);
- }
-
- if (packet is MqttSubscribePacket subscribePacket)
- {
- return HandleIncomingSubscribePacketAsync(adapter, subscribePacket, cancellationToken);
- }
-
- if (packet is MqttUnsubscribePacket unsubscribePacket)
- {
- return HandleIncomingUnsubscribePacketAsync(adapter, unsubscribePacket, cancellationToken);
- }
-
- if (packet is MqttDisconnectPacket)
- {
- return StopAsync(MqttClientDisconnectType.Clean);
- }
-
- if (packet is MqttConnectPacket)
- {
- return StopAsync(MqttClientDisconnectType.NotClean);
- }
-
- _logger.Warning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
- return StopAsync(MqttClientDisconnectType.NotClean);
- }
-
- private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
- {
- var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
- await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { subscribeResult.ResponsePacket }).ConfigureAwait(false);
-
- if (subscribeResult.CloseConnection)
- {
- await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
- }
-
- await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
- }
-
- private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
- {
- var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
- await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { unsubscribeResult });
- }
-
- private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection topicFilters)
- {
- var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters);
- foreach (var applicationMessage in retainedMessages)
- {
- await EnqueueApplicationMessageAsync(applicationMessage).ConfigureAwait(false);
- }
- }
-
- private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
- {
- var applicationMessage = publishPacket.ToApplicationMessage();
-
- switch (applicationMessage.QualityOfServiceLevel)
- {
- case MqttQualityOfServiceLevel.AtMostOnce:
- {
- return ApplicationMessageReceivedCallback?.Invoke(this, applicationMessage);
- }
- case MqttQualityOfServiceLevel.AtLeastOnce:
- {
- return HandleIncomingPublishPacketWithQoS1(adapter, applicationMessage, publishPacket, cancellationToken);
- }
- case MqttQualityOfServiceLevel.ExactlyOnce:
- {
- return HandleIncomingPublishPacketWithQoS2(adapter, applicationMessage, publishPacket, cancellationToken);
- }
- default:
- {
- throw new MqttCommunicationException("Received a not supported QoS level.");
- }
- }
- }
-
- private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
- {
- await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
-
- var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
- 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 ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
-
- var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
- 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, new[] { response });
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs
deleted file mode 100644
index 5f6cf15e..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs
+++ /dev/null
@@ -1,335 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
-using MQTTnet.Exceptions;
-using MQTTnet.Packets;
-using MQTTnet.Protocol;
-using MQTTnet.Serializer;
-
-namespace MQTTnet.Server
-{
- public sealed class MqttClientSessionsManager : IDisposable
- {
- private readonly Dictionary _sessions = new Dictionary();
- private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
-
- private readonly IMqttServerOptions _options;
- private readonly MqttRetainedMessagesManager _retainedMessagesManager;
- private readonly IMqttNetLogger _logger;
-
- public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
- {
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
- _options = options ?? throw new ArgumentNullException(nameof(options));
- _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
- }
-
- public Action ClientConnectedCallback { get; set; }
- public Action ClientDisconnectedCallback { get; set; }
- public Action ClientSubscribedTopicCallback { get; set; }
- public Action ClientUnsubscribedTopicCallback { get; set; }
- public Action ApplicationMessageReceivedCallback { get; set; }
-
- public async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
- {
- var clientId = string.Empty;
- var wasCleanDisconnect = false;
- MqttClientSession clientSession = null;
-
- try
- {
- 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].");
- }
-
- clientId = connectPacket.ClientId;
-
- // Switch to the required protocol version before sending any response.
- clientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion;
-
- var connectReturnCode = ValidateConnection(connectPacket);
- if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
- {
- await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
- {
- new MqttConnAckPacket
- {
- ConnectReturnCode = connectReturnCode
- }
- }).ConfigureAwait(false);
-
- return;
- }
-
- var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
- clientSession = result.Session;
-
- await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
- {
- new MqttConnAckPacket
- {
- ConnectReturnCode = connectReturnCode,
- IsSessionPresent = result.IsExistingSession
- }
- }).ConfigureAwait(false);
-
- ClientConnectedCallback?.Invoke(new ConnectedMqttClient
- {
- ClientId = clientId,
- ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
- });
-
- wasCleanDisconnect = await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
- }
- catch (OperationCanceledException)
- {
- }
- catch (Exception exception)
- {
- _logger.Error(exception, exception.Message);
- }
- finally
- {
- try
- {
- await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
- clientAdapter.Dispose();
- }
- catch (Exception exception)
- {
- _logger.Error(exception, exception.Message);
- }
-
- ClientDisconnectedCallback?.Invoke(new ConnectedMqttClient
- {
- ClientId = clientId,
- ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion,
- PendingApplicationMessages = clientSession?.PendingMessagesQueue.Count ?? 0
- },
- wasCleanDisconnect);
- }
- }
-
- public async Task StopAsync()
- {
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- foreach (var session in _sessions)
- {
- await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
- }
-
- _sessions.Clear();
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- public async Task> GetConnectedClientsAsync()
- {
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
- {
- ClientId = s.Value.ClientId,
- ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311,
- LastPacketReceived = s.Value.KeepAliveMonitor.LastPacketReceived,
- LastNonKeepAlivePacketReceived = s.Value.KeepAliveMonitor.LastNonKeepAlivePacketReceived,
- PendingApplicationMessages = s.Value.PendingMessagesQueue.Count
- }).ToList();
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- public async Task DispatchApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
- {
- try
- {
- var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
- if (interceptorContext.CloseConnection)
- {
- await senderClientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
- }
-
- if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
- {
- return;
- }
-
- if (applicationMessage.Retain)
- {
- await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
- }
-
- ApplicationMessageReceivedCallback?.Invoke(senderClientSession?.ClientId, applicationMessage);
- }
- catch (Exception exception)
- {
- _logger.Error(exception, "Error while processing application message");
- }
-
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- foreach (var clientSession in _sessions.Values)
- {
- await clientSession.EnqueueApplicationMessageAsync(applicationMessage);
- }
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- public async Task SubscribeAsync(string clientId, IList 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 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 MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
- {
- var interceptorContext = new MqttApplicationMessageInterceptorContext(
- senderClientSession?.ClientId,
- applicationMessage);
-
- var interceptor = _options.ApplicationMessageInterceptor;
- if (interceptor == null)
- {
- return interceptorContext;
- }
-
- interceptor(interceptorContext);
- return interceptorContext;
- }
-
- private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
- {
- if (_options.ConnectionValidator == null)
- {
- return MqttConnectReturnCode.ConnectionAccepted;
- }
-
- var context = new MqttConnectionValidatorContext(
- connectPacket.ClientId,
- connectPacket.Username,
- connectPacket.Password,
- connectPacket.WillMessage);
-
- _options.ConnectionValidator(context);
- return context.ReturnCode;
- }
-
- private async Task GetOrCreateClientSessionAsync(MqttConnectPacket connectPacket)
- {
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
- if (isSessionPresent)
- {
- if (connectPacket.CleanSession)
- {
- _sessions.Remove(connectPacket.ClientId);
-
- await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false);
- clientSession.Dispose();
- clientSession = null;
-
- _logger.Verbose("Stopped existing session of client '{0}'.", connectPacket.ClientId);
- }
- else
- {
- _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
- }
- }
-
- var isExistingSession = true;
- if (clientSession == null)
- {
- isExistingSession = false;
-
- clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, _logger)
- {
- ApplicationMessageReceivedCallback = DispatchApplicationMessageAsync
- };
-
- clientSession.SubscriptionsManager.TopicSubscribedCallback = ClientSubscribedTopicCallback;
- clientSession.SubscriptionsManager.TopicUnsubscribedCallback = ClientUnsubscribedTopicCallback;
-
- _sessions[connectPacket.ClientId] = clientSession;
-
- _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
- }
-
- return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- public void Dispose()
- {
- ClientConnectedCallback = null;
- ClientDisconnectedCallback = null;
- ClientSubscribedTopicCallback = null;
- ClientUnsubscribedTopicCallback = null;
- ApplicationMessageReceivedCallback = null;
-
- _semaphore?.Dispose();
- }
- }
-}
\ No newline at end of file
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs
deleted file mode 100644
index f998557e..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs
+++ /dev/null
@@ -1,144 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
-
-namespace MQTTnet.Server
-{
- public sealed class MqttRetainedMessagesManager : IDisposable
- {
- private readonly Dictionary _retainedMessages = new Dictionary();
- private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
- private readonly IMqttNetLogger _logger;
- private readonly IMqttServerOptions _options;
-
- public MqttRetainedMessagesManager(IMqttServerOptions options, IMqttNetLogger logger)
- {
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
- _options = options ?? throw new ArgumentNullException(nameof(options));
- }
-
- public async Task LoadMessagesAsync()
- {
- if (_options.Storage == null)
- {
- return;
- }
-
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync();
-
- _retainedMessages.Clear();
- foreach (var retainedMessage in retainedMessages)
- {
- _retainedMessages[retainedMessage.Topic] = retainedMessage;
- }
- }
- catch (Exception exception)
- {
- _logger.Error(exception, "Unhandled exception while loading retained messages.");
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- public async Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage)
- {
- if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- await HandleMessageInternalAsync(clientId, applicationMessage);
- }
- catch (Exception exception)
- {
- _logger.Error(exception, "Unhandled exception while handling retained messages.");
- }
- finally
- {
- _semaphore.Release();
- }
- }
-
- public async Task> GetSubscribedMessagesAsync(ICollection topicFilters)
- {
- var retainedMessages = new List();
-
- await _semaphore.WaitAsync().ConfigureAwait(false);
- try
- {
- foreach (var retainedMessage in _retainedMessages.Values)
- {
- foreach (var topicFilter in topicFilters)
- {
- if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic))
- {
- continue;
- }
-
- retainedMessages.Add(retainedMessage);
- break;
- }
- }
- }
- finally
- {
- _semaphore.Release();
- }
-
- return retainedMessages;
- }
-
- public void Dispose()
- {
- _semaphore?.Dispose();
- }
-
- private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
- {
- var saveIsRequired = false;
-
- if (applicationMessage.Payload?.Any() == false)
- {
- saveIsRequired = _retainedMessages.Remove(applicationMessage.Topic);
- _logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
- }
- else
- {
- if (!_retainedMessages.ContainsKey(applicationMessage.Topic))
- {
- _retainedMessages[applicationMessage.Topic] = applicationMessage;
- saveIsRequired = true;
- }
- else
- {
- var existingMessage = _retainedMessages[applicationMessage.Topic];
- if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
- {
- _retainedMessages[applicationMessage.Topic] = applicationMessage;
- saveIsRequired = true;
- }
- }
-
- _logger.Info("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
- }
-
- if (!saveIsRequired)
- {
- _logger.Verbose("Skipped saving retained messages because no changes were detected.");
- }
-
- if (saveIsRequired && _options.Storage != null)
- {
- await _options.Storage.SaveRetainedMessagesAsync(_retainedMessages.Values.ToList());
- }
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttServerDefaultEndpointOptions.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttServerDefaultEndpointOptions.cs
deleted file mode 100644
index 545499b5..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttServerDefaultEndpointOptions.cs
+++ /dev/null
@@ -1,13 +0,0 @@
-using System.Net;
-
-namespace MQTTnet.Server
-{
- public sealed class MqttServerDefaultEndpointOptions
- {
- public bool IsEnabled { get; set; } = true;
-
- public int? Port { get; set; }
-
- public IPAddress BoundIPAddress { get; set; } = IPAddress.Any;
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttServerOptions.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttServerOptions.cs
deleted file mode 100644
index 2b0a9a00..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttServerOptions.cs
+++ /dev/null
@@ -1,23 +0,0 @@
-using System;
-
-namespace MQTTnet.Server
-{
- public class MqttServerOptions : IMqttServerOptions
- {
- public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions();
-
- public MqttServerTlsEndpointOptions TlsEndpointOptions { get; } = new MqttServerTlsEndpointOptions();
-
- public int ConnectionBacklog { get; set; } = 10;
-
- public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
-
- public Action ConnectionValidator { get; set; }
-
- public Action ApplicationMessageInterceptor { get; set; }
-
- public Action SubscriptionInterceptor { get; set; }
-
- public IMqttServerStorage Storage { get; set; }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttServerOptionsExtensions.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttServerOptionsExtensions.cs
deleted file mode 100644
index db93ef18..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttServerOptionsExtensions.cs
+++ /dev/null
@@ -1,31 +0,0 @@
-using System;
-
-namespace MQTTnet.Server
-{
- public static class MqttServerOptionsExtensions
- {
- public static int GetTlsEndpointPort(this IMqttServerOptions options)
- {
- if (options == null) throw new ArgumentNullException(nameof(options));
-
- if (!options.TlsEndpointOptions.Port.HasValue)
- {
- return 8883;
- }
-
- return options.TlsEndpointOptions.Port.Value;
- }
-
- public static int GetDefaultEndpointPort(this IMqttServerOptions options)
- {
- if (options == null) throw new ArgumentNullException(nameof(options));
-
- if (!options.DefaultEndpointOptions.Port.HasValue)
- {
- return 1883;
- }
-
- return options.DefaultEndpointOptions.Port.Value;
- }
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttServerStartedEventArgs.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttServerStartedEventArgs.cs
deleted file mode 100644
index 08202994..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttServerStartedEventArgs.cs
+++ /dev/null
@@ -1,8 +0,0 @@
-using System;
-
-namespace MQTTnet.Server
-{
- public class MqttServerStartedEventArgs : EventArgs
- {
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttServerTlsEndpointOptions.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttServerTlsEndpointOptions.cs
deleted file mode 100644
index 73be93c5..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttServerTlsEndpointOptions.cs
+++ /dev/null
@@ -1,15 +0,0 @@
-using System.Net;
-
-namespace MQTTnet.Server
-{
- public sealed class MqttServerTlsEndpointOptions
- {
- public bool IsEnabled { get; set; }
-
- public int? Port { get; set; }
-
- public byte[] Certificate { get; set; }
-
- public IPAddress BoundIPAddress { get; set; } = IPAddress.Any;
- }
-}
diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs
deleted file mode 100644
index 82444159..00000000
--- a/Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs
+++ /dev/null
@@ -1,44 +0,0 @@
-using System;
-
-namespace MQTTnet.Server
-{
- public static class MqttTopicFilterComparer
- {
- private static readonly char[] TopicLevelSeparator = { '/' };
-
- public static bool IsMatch(string topic, string filter)
- {
- if (topic == null) throw new ArgumentNullException(nameof(topic));
- if (filter == null) throw new ArgumentNullException(nameof(filter));
-
- if (string.Equals(topic, filter, StringComparison.Ordinal))
- {
- return true;
- }
-
- var fragmentsTopic = topic.Split(TopicLevelSeparator, StringSplitOptions.None);
- var fragmentsFilter = filter.Split(TopicLevelSeparator, StringSplitOptions.None);
-
- for (var i = 0; i < fragmentsFilter.Length; i++)
- {
- switch (fragmentsFilter[i])
- {
- case "+": continue;
- case "#" when i == fragmentsFilter.Length - 1: return true;
- }
-
- if (i >= fragmentsTopic.Length)
- {
- return false;
- }
-
- if (!string.Equals(fragmentsFilter[i], fragmentsTopic[i], StringComparison.Ordinal))
- {
- return false;
- }
- }
-
- return fragmentsTopic.Length <= fragmentsFilter.Length;
- }
- }
-}
diff --git a/MQTTnet.sln b/MQTTnet.sln
index b60e21d5..be6a0b6f 100644
--- a/MQTTnet.sln
+++ b/MQTTnet.sln
@@ -7,23 +7,26 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Core.Tests", "Tests
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Tests", "Tests", "{9248C2E1-B9D6-40BF-81EC-86004D7765B4}"
EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Frameworks", "Frameworks", "{32A630A7-2598-41D7-B625-204CD906F5FB}"
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Source", "Source", "{32A630A7-2598-41D7-B625-204CD906F5FB}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.TestApp.UniversalWindows", "Tests\MQTTnet.TestApp.UniversalWindows\MQTTnet.TestApp.UniversalWindows.csproj", "{FF1F72D6-9524-4422-9497-3CC0002216ED}"
EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.NetStandard", "Frameworks\MQTTnet.NetStandard\MQTTnet.Netstandard.csproj", "{3587E506-55A2-4EB3-99C7-DC01E42D25D2}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet", "Source\MQTTnet\MQTTnet.csproj", "{3587E506-55A2-4EB3-99C7-DC01E42D25D2}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1-BC3A-420A-BE9C-FA2401431CF9}"
ProjectSection(SolutionItems) = preProject
- appveyor.yml = appveyor.yml
Build\build.ps1 = Build\build.ps1
Build\MQTTnet.AspNetCore.nuspec = Build\MQTTnet.AspNetCore.nuspec
+ Build\MQTTnet.Extensions.ManagedClient.nuspec = Build\MQTTnet.Extensions.ManagedClient.nuspec
Build\MQTTnet.Extensions.Rpc.nuspec = Build\MQTTnet.Extensions.Rpc.nuspec
Build\MQTTnet.nuspec = Build\MQTTnet.nuspec
+ Build\upload.ps1 = Build\upload.ps1
EndProjectSection
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}"
ProjectSection(SolutionItems) = preProject
+ .bettercodehub.yml = .bettercodehub.yml
+ appveyor.yml = appveyor.yml
LICENSE = LICENSE
README.md = README.md
EndProjectSection
@@ -32,13 +35,17 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.TestApp.NetCore", "
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.TestApp.AspNetCore2", "Tests\MQTTnet.TestApp.AspNetCore2\MQTTnet.TestApp.AspNetCore2.csproj", "{C6FF8AEA-0855-41EC-A1F3-AC262225BAB9}"
EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Frameworks\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Extensions", "Extensions", "{12816BCC-AF9E-44A9-9AE5-C246AF2A0587}"
EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.Rpc", "Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj", "{C444E9C8-95FA-430E-9126-274129DE16CD}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.Rpc", "Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj", "{C444E9C8-95FA-430E-9126-274129DE16CD}"
EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Benchmarks", "Tests\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{998D04DD-7CB0-45F5-A393-E2495C16399E}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Benchmarks", "Tests\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{998D04DD-7CB0-45F5-A393-E2495C16399E}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.ManagedClient", "Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj", "{C400533A-8EBA-4F0B-BF4D-295C3708604B}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore.Tests", "Tests\MQTTnet.AspNetCore.Tests\MQTTnet.AspNetCore.Tests.csproj", "{61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -70,6 +77,7 @@ Global
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.Build.0 = Release|Any CPU
{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|Any CPU.Deploy.0 = Debug|Any CPU
{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|ARM.ActiveCfg = Debug|ARM
{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|ARM.Build.0 = Debug|ARM
{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|ARM.Deploy.0 = Debug|ARM
@@ -186,6 +194,38 @@ Global
{998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x64.Build.0 = Release|Any CPU
{998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x86.ActiveCfg = Release|Any CPU
{998D04DD-7CB0-45F5-A393-E2495C16399E}.Release|x86.Build.0 = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|ARM.ActiveCfg = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|ARM.Build.0 = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|x64.Build.0 = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Debug|x86.Build.0 = Debug|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|Any CPU.Build.0 = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|ARM.ActiveCfg = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|ARM.Build.0 = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|x64.ActiveCfg = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|x64.Build.0 = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|x86.ActiveCfg = Release|Any CPU
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B}.Release|x86.Build.0 = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|ARM.ActiveCfg = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|ARM.Build.0 = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|x64.Build.0 = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Debug|x86.Build.0 = Debug|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|Any CPU.Build.0 = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|ARM.ActiveCfg = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|ARM.Build.0 = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|x64.ActiveCfg = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|x64.Build.0 = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|x86.ActiveCfg = Release|Any CPU
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E}.Release|x86.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@@ -199,6 +239,8 @@ Global
{F10C4060-F7EE-4A83-919F-FF723E72F94A} = {32A630A7-2598-41D7-B625-204CD906F5FB}
{C444E9C8-95FA-430E-9126-274129DE16CD} = {12816BCC-AF9E-44A9-9AE5-C246AF2A0587}
{998D04DD-7CB0-45F5-A393-E2495C16399E} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4}
+ {C400533A-8EBA-4F0B-BF4D-295C3708604B} = {12816BCC-AF9E-44A9-9AE5-C246AF2A0587}
+ {61B62223-F5D0-48E4-BBD6-2CBA9353CB5E} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894}
diff --git a/README.md b/README.md
index 6a3111d8..7181d724 100644
--- a/README.md
+++ b/README.md
@@ -5,6 +5,8 @@
[](https://www.nuget.org/packages/MQTTnet)
[](https://ci.appveyor.com/project/chkr1011/mqttnet/branch/develop)
[](https://bettercodehub.com/)
+[](https://opencollective.com/mqttnet)
+[](https://opencollective.com/mqttnet)
# MQTTnet
@@ -18,10 +20,10 @@ 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 ~40.000 messages / second)*
+* Performance optimized (processing ~70.000 messages / second)*
* Interfaces included for mocking and testing
* Access to internal trace messages
-* Unit tested (70+ tests)
+* Unit tested (~90 tests)
\* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
@@ -50,14 +52,15 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
* .NET Standard 1.3+
* .NET Core 1.1+
* .NET Core App 1.1+
-* Universal Windows Platform (UWP) 10.0.10240+ (x86, x64, ARM, AnyCPU, Windows 10 IoT Core)
* .NET Framework 4.5.2+ (x86, x64, AnyCPU)
* Mono 5.2+
+* Universal Windows Platform (UWP) 10.0.10240+ (x86, x64, ARM, AnyCPU, Windows 10 IoT Core)
* Xamarin.Android 7.5+
* Xamarin.iOS 10.14+
## Supported MQTT versions
+* 5.0.0 (planned)
* 3.1.1
* 3.1.0
@@ -72,6 +75,9 @@ Please find examples and the documentation at the Wiki of this repository ()
* MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester) and [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8))
-* Wirehome (Open Source Home Automation system for .NET, )
-
+* HA4IoT (Open Source Home Automation system for .NET, )
If you use this library and want to see your project here please let me know.
## MIT License
-Copyright (c) 2017 Christian Kratky
+Copyright (c) 2017-2018 Christian Kratky
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
diff --git a/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
new file mode 100644
index 00000000..22b9e937
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
@@ -0,0 +1,52 @@
+using System;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.Extensions.DependencyInjection;
+using System.Linq;
+using MQTTnet.Server;
+
+namespace MQTTnet.AspNetCore
+{
+ public static class ApplicationBuilderExtensions
+ {
+ public static IApplicationBuilder UseMqttEndpoint(this IApplicationBuilder app, string path = "/mqtt")
+ {
+ app.UseWebSockets();
+ app.Use(async (context, next) =>
+ {
+ if (!context.WebSockets.IsWebSocketRequest || context.Request.Path != path)
+ {
+ await next();
+ return;
+ }
+
+ string subProtocol = null;
+
+ if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues))
+ {
+ // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
+ subProtocol = requestedSubProtocolValues
+ .OrderByDescending(p => p.Length)
+ .FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
+ }
+
+ var adapter = app.ApplicationServices.GetRequiredService();
+ using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subProtocol))
+ {
+ var endpoint = $"{context.Connection.RemoteIpAddress}:{context.Connection.RemotePort}";
+ await adapter.RunWebSocketConnectionAsync(webSocket, endpoint);
+ }
+ });
+
+ return app;
+ }
+
+ public static IApplicationBuilder UseMqttServer(this IApplicationBuilder app, Action configure)
+ {
+ var server = app.ApplicationServices.GetRequiredService();
+
+ configure(server);
+
+ return app;
+ }
+ }
+}
diff --git a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs
new file mode 100644
index 00000000..e308ccd4
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs
@@ -0,0 +1,34 @@
+using System;
+using System.Net;
+using MQTTnet.Adapter;
+using MQTTnet.AspNetCore.Client.Tcp;
+using MQTTnet.Client;
+using MQTTnet.Diagnostics;
+using MQTTnet.Serializer;
+
+namespace MQTTnet.AspNetCore.Client
+{
+ public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
+ {
+ public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
+ {
+ if (options == null) throw new ArgumentNullException(nameof(options));
+
+ var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion };
+
+ switch (options.ChannelOptions)
+ {
+ case MqttClientTcpOptions tcpOptions:
+ {
+ var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
+ var tcpConnection = new TcpConnection(endpoint);
+ return new MqttConnectionContext(serializer, tcpConnection);
+ }
+ default:
+ {
+ throw new NotSupportedException();
+ }
+ }
+ }
+ }
+}
diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs
new file mode 100644
index 00000000..5911a3a1
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs
@@ -0,0 +1,22 @@
+using System;
+using System.Runtime.InteropServices;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+ public static class BufferExtensions
+ {
+ public static ArraySegment GetArray(this Memory memory)
+ {
+ return ((ReadOnlyMemory)memory).GetArray();
+ }
+
+ public static ArraySegment GetArray(this ReadOnlyMemory memory)
+ {
+ if (!MemoryMarshal.TryGetArray(memory, out var result))
+ {
+ throw new InvalidOperationException("Buffer backed by array was expected");
+ }
+ return result;
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs
new file mode 100644
index 00000000..e234da55
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs
@@ -0,0 +1,41 @@
+using System.IO.Pipelines;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+ public class DuplexPipe : IDuplexPipe
+ {
+ public DuplexPipe(PipeReader reader, PipeWriter writer)
+ {
+ Input = reader;
+ Output = writer;
+ }
+
+ public PipeReader Input { get; }
+
+ public PipeWriter Output { get; }
+
+ public static DuplexPipePair CreateConnectionPair(PipeOptions inputOptions, PipeOptions outputOptions)
+ {
+ var input = new Pipe(inputOptions);
+ var output = new Pipe(outputOptions);
+
+ var transportToApplication = new DuplexPipe(output.Reader, input.Writer);
+ var applicationToTransport = new DuplexPipe(input.Reader, output.Writer);
+
+ return new DuplexPipePair(applicationToTransport, transportToApplication);
+ }
+
+ // This class exists to work around issues with value tuple on .NET Framework
+ public readonly struct DuplexPipePair
+ {
+ public IDuplexPipe Transport { get; }
+ public IDuplexPipe Application { get; }
+
+ public DuplexPipePair(IDuplexPipe transport, IDuplexPipe application)
+ {
+ Transport = transport;
+ Application = application;
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs
new file mode 100644
index 00000000..dbc26120
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs
@@ -0,0 +1,70 @@
+using System;
+using System.Diagnostics;
+using System.IO.Pipelines;
+using System.Net.Sockets;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+ public class SocketAwaitable : ICriticalNotifyCompletion
+ {
+ private static readonly Action _callbackCompleted = () => { };
+
+ private readonly PipeScheduler _ioScheduler;
+
+ private Action _callback;
+ private int _bytesTransferred;
+ private SocketError _error;
+
+ public SocketAwaitable(PipeScheduler ioScheduler)
+ {
+ _ioScheduler = ioScheduler;
+ }
+
+ public bool IsCompleted => ReferenceEquals(_callback, _callbackCompleted);
+
+ public SocketAwaitable GetAwaiter() => this;
+
+ public int GetResult()
+ {
+ Debug.Assert(ReferenceEquals(_callback, _callbackCompleted));
+
+ _callback = null;
+
+ if (_error != SocketError.Success)
+ {
+ throw new SocketException((int)_error);
+ }
+
+ return _bytesTransferred;
+ }
+
+ public void OnCompleted(Action continuation)
+ {
+ if (ReferenceEquals(_callback, _callbackCompleted) ||
+ ReferenceEquals(Interlocked.CompareExchange(ref _callback, continuation, null), _callbackCompleted))
+ {
+ Task.Run(continuation);
+ }
+ }
+
+ public void UnsafeOnCompleted(Action continuation)
+ {
+ OnCompleted(continuation);
+ }
+
+ public void Complete(int bytesTransferred, SocketError socketError)
+ {
+ _error = socketError;
+ _bytesTransferred = bytesTransferred;
+ var continuation = Interlocked.Exchange(ref _callback, _callbackCompleted);
+
+ if (continuation != null)
+ {
+ _ioScheduler.Schedule(state => ((Action)state)(), continuation);
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs
new file mode 100644
index 00000000..7d11fa2b
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs
@@ -0,0 +1,37 @@
+using System;
+using System.IO.Pipelines;
+using System.Net.Sockets;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+ public class SocketReceiver
+ {
+ private readonly Socket _socket;
+ private readonly SocketAsyncEventArgs _eventArgs = new SocketAsyncEventArgs();
+ private readonly SocketAwaitable _awaitable;
+
+ public SocketReceiver(Socket socket, PipeScheduler scheduler)
+ {
+ _socket = socket;
+ _awaitable = new SocketAwaitable(scheduler);
+ _eventArgs.UserToken = _awaitable;
+ _eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
+ }
+
+ public SocketAwaitable ReceiveAsync(Memory buffer)
+ {
+#if NETCOREAPP2_1
+ _eventArgs.SetBuffer(buffer);
+#else
+ var segment = buffer.GetArray();
+ _eventArgs.SetBuffer(segment.Array, segment.Offset, segment.Count);
+#endif
+ if (!_socket.ReceiveAsync(_eventArgs))
+ {
+ _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
+ }
+
+ return _awaitable;
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs
new file mode 100644
index 00000000..55192d6c
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs
@@ -0,0 +1,97 @@
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO.Pipelines;
+using System.Net.Sockets;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+ public class SocketSender
+ {
+ private readonly Socket _socket;
+ private readonly SocketAsyncEventArgs _eventArgs = new SocketAsyncEventArgs();
+ private readonly SocketAwaitable _awaitable;
+
+ private List> _bufferList;
+
+ public SocketSender(Socket socket, PipeScheduler scheduler)
+ {
+ _socket = socket;
+ _awaitable = new SocketAwaitable(scheduler);
+ _eventArgs.UserToken = _awaitable;
+ _eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
+ }
+
+ public SocketAwaitable SendAsync(in ReadOnlySequence buffers)
+ {
+ if (buffers.IsSingleSegment)
+ {
+ return SendAsync(buffers.First);
+ }
+
+#if NETCOREAPP2_1
+ if (!_eventArgs.MemoryBuffer.Equals(Memory.Empty))
+#else
+ if (_eventArgs.Buffer != null)
+#endif
+ {
+ _eventArgs.SetBuffer(null, 0, 0);
+ }
+
+ _eventArgs.BufferList = GetBufferList(buffers);
+
+ if (!_socket.SendAsync(_eventArgs))
+ {
+ _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
+ }
+
+ return _awaitable;
+ }
+
+ private SocketAwaitable SendAsync(ReadOnlyMemory memory)
+ {
+ // The BufferList getter is much less expensive then the setter.
+ if (_eventArgs.BufferList != null)
+ {
+ _eventArgs.BufferList = null;
+ }
+
+#if NETCOREAPP2_1
+ _eventArgs.SetBuffer(MemoryMarshal.AsMemory(memory));
+#else
+ var segment = memory.GetArray();
+ _eventArgs.SetBuffer(segment.Array, segment.Offset, segment.Count);
+#endif
+ if (!_socket.SendAsync(_eventArgs))
+ {
+ _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
+ }
+
+ return _awaitable;
+ }
+
+ private List> GetBufferList(in ReadOnlySequence buffer)
+ {
+ Debug.Assert(!buffer.IsEmpty);
+ Debug.Assert(!buffer.IsSingleSegment);
+
+ if (_bufferList == null)
+ {
+ _bufferList = new List>();
+ }
+ else
+ {
+ // Buffers are pooled, so it's OK to root them until the next multi-buffer write.
+ _bufferList.Clear();
+ }
+
+ foreach (var b in buffer)
+ {
+ _bufferList.Add(b.GetArray());
+ }
+
+ return _bufferList;
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs
new file mode 100644
index 00000000..37913fe4
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs
@@ -0,0 +1,258 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipelines;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Http.Features;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+ public class TcpConnection : ConnectionContext
+ {
+ private volatile bool _aborted;
+ private readonly EndPoint _endPoint;
+ private SocketSender _sender;
+ private SocketReceiver _receiver;
+
+ private Socket _socket;
+ private IDuplexPipe _application;
+
+ public bool IsConnected { get; private set; }
+ public override string ConnectionId { get; set; }
+ public override IFeatureCollection Features { get; }
+ public override IDictionary