Compare commits

...
Author SHA1 Message Date
Christian 05dcb250e7 Add README to nugets 2024-04-20 14:23:15 +02:00
christian 7f5c437c46 Fix unit tests 2024-04-11 22:26:39 +02:00
christian ae5e038548 Merge remote-tracking branch 'origin/master' 2024-04-09 21:34:31 +02:00
christian 67e0fb8b0b Fix build errors by no longer using not supported language features for certain TFMs. 2024-04-09 21:34:15 +02:00
Marc PiulachsandChristian b763542329 Move "Client is not connected" error to it's own exception (#1954)
* Move "Client is not connected" error to it's own exception

* Update MqttClientNotConnectedException.cs

* Update release notes

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2024-04-09 21:26:59 +02:00
christian d061da6077 Update release notes 2024-04-09 21:24:39 +02:00
Andriy Tolstoy 82a235aef4 remove select method (#1959) 2024-04-09 21:18:09 +02:00
James Thompson fd55982d98 #1957 Invert the logic to improve maintainance (#1958)
#1957 Invert the logic to improve maintaince
2024-04-01 22:13:48 +02:00
aabf180b23 1862 Add subscribe and unsubscribe results event for Managed Mqtt Client. (#1894)
* See the issue #1862

Adding an event when ManagedMqttClient actually subscribe/unsubscribe to topics.
This event has to be used in parallel with connected event to know the exact state.

* Rename SubscriptionsResultAsync to SubscriptionsChangedAsync.

* Rename internal field for event

* Update ReleaseNotes.md

---------

Co-authored-by: Pierre Dufrene <pierre.dufrene@se.com>
Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2024-03-02 09:58:42 +01:00
Christian 1245d8299a Merge branch 'master' of https://github.com/dotnet/MQTTnet 2024-03-02 08:45:03 +01:00
yztandChristian 1ffffaff47 Fix minor bug in MqttV5PacketEncoder.cs (#1939)
* Fix minor bug in MqttV5PacketEncoder.cs

* Update ReleaseNotes.md

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2024-03-02 09:44:14 +01:00
Christian c204f6d8b2 Make packet inspector fully async (#1941)
* Make packet inspector methods async.

* Update release notes
2024-03-02 09:28:56 +01:00
Christian b775fb256b Update version 2024-03-02 08:44:57 +01:00
Christian f82d382f23 Merge remote-tracking branch 'origin/master' 2024-02-14 17:54:04 +01:00
Christian 4214af283a Update ReleaseNotes.md 2024-02-14 17:47:29 +01:00
Christian b32a5c9fa0 Update ReleaseNotes.md 2024-02-14 17:44:16 +01:00
Sean Hanna e7e13a631f implement intercept publish message event on ManagedMqttClient (#1915)
* implement intercept publish message event on ManagedMqttClient

* rename ShouldPublish to ACceptPublish per PR feedback
2024-02-14 17:43:45 +01:00
Christian c12096705f 1908 support unix domain sockets (#1919)
* Add support for custom endpoints

Add overlaods etc. to allow for passing other endpoint implementations

* Extend publich broker test to also probe with different protocol versions

* Fix UWP build

* Update ReleaseNotes.md
2024-02-06 17:37:07 +01:00
Christian eb25500f50 1889 no service for type mqttnetservermqttserver has been registered (#1892)
* Add missing service to service collection

* Update ReleaseNotes.md

* Restore service registrations
2023-12-08 18:01:19 +01:00
Christian 4df06980a1 1883 in a weak network environment the mqtt broker is unable to detect disconnections (#1891)
* Add new setting

* Align keep alive handling with RFC

Only completely sent packets are treated as keep alive

* Update ReleaseNotes.md
2023-12-07 17:49:03 +01:00
Christian f3499013f4 Increase version 2023-12-07 17:48:14 +01:00
Christian 158a7fca46 Reset release notes 2023-12-07 17:47:57 +01:00
kallayjandChristian 09ab30fb0f Adds events for delivered and dropped messages. (#1866)
* Adds events for delivered and dropped messages.

* Rename new events to match other namings

* Update ReleaseNotes.md

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2023-12-06 18:02:48 +01:00
Christian ae84aa43d2 1846 server mqttserverstopasync doesnt indicate correct reason (#1872) 2023-11-25 16:17:16 +01:00
RidoandChristian b431a5d3fe Add ReasonCode to SendExtendedAuthenticationExchangeDataAsync (#1882)
* add reason code to ext auth

* Update ReleaseNotes.md

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2023-11-25 15:45:10 +01:00
Christian b9683c2943 Improve client disconnect (#1873)
Imrpove client disconnect
2023-11-05 15:15:36 +01:00
Christian 1ba8f74555 Update version 2023-11-04 13:40:28 +01:00
Christian ec637f7946 Handle unobserved tasks exceptions (#1871)
* Access exception so that it will not treated as unhandled

* Refactor code

* Improve unit tests

* Fix null reference exception

* Update ReleaseNotes.md
2023-11-04 13:39:16 +01:00
Christian 2599b3dfdc Update settings for tests project so that it will build on linux (#1860)
* Update settings for tests project so that it will build on linux

* Improve unit tests and samples

* Remove no longer required target frameworks for unit tests

* Update ci.yml
2023-11-01 09:34:18 +01:00
Christian d22caa7b70 Update readme (#1870)
* Update ReleaseNotes.md

* Apply project code style
2023-11-01 09:26:18 +01:00
Rido 01c90badf9 Support custom CA chain validation (#1851)
* Add CaFile based on SslStream CustomTrust

* change to cert collection

* do not require to disable revocation checks
2023-11-01 09:22:13 +01:00
Diego2098andChristian e4f605fbd5 Update MqttServer.cs (#1858)
* Update MqttServer.cs

clientID is null => RetainedMessageChangedEventArgs ctor throw new ArgumentNullException(nameof(clientId)

* Update ReleaseNotes.md

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2023-11-01 08:29:19 +01:00
85 changed files with 1686 additions and 651 deletions
+4 -2
View File
@@ -1,2 +1,4 @@
* [Client] Fixed wrong TLS options handling (#1830).
* [Client] Fixed NullReferenceExeption when performing a Ping when the client is not connected (#1831).
* [Core] Optimized packet serialization of PUBACK and PUBREC packets for protocol version 5.0.0 (#1939, thanks to @Y-Sindo).
* [Core] The package inspector is now fully async (#1941).
* [Client] Added a dedicated exception when the client is not connected (#1954, thanks to @marcpiulachs).
* [ManagedClient] Added a new event (SubscriptionsChangedAsync) which is fired when a subscription or unsubscription was made (#1894, thanks to @pdufrene).
+2 -2
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request]
env:
VERSION: "4.3.1.${{github.run_number}}"
VERSION: "4.3.4.${{github.run_number}}"
jobs:
build:
@@ -54,7 +54,7 @@ jobs:
uses: darenm/Setup-VSTest@v1
- name: Core Tests
run: vstest.console.exe Source\MQTTnet.Tests\bin\Release\net6.0\MQTTnet.Tests.dll
run: vstest.console.exe Source\MQTTnet.Tests\bin\Release\net7.0\MQTTnet.Tests.dll
- name: ASP.NET Tests
run: vstest.console.exe Source\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll
+2
View File
@@ -230,6 +230,7 @@
<s:String x:Key="/Default/CodeStyle/FileHeader/FileHeaderText/@EntryValue">Licensed to the .NET Foundation under one or more agreements.
The .NET Foundation licenses this file to you under the MIT license.
See the LICENSE file in the project root for more information.</s:String>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EFeature_002EServices_002ECodeCleanup_002EFileHeader_002EFileHeaderSettingsMigrate/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
@@ -240,6 +241,7 @@ See the LICENSE file in the project root for more information.</s:String>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=catched/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=CONNACK/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean>
@@ -7,6 +7,7 @@
// ReSharper disable InconsistentNaming
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using MQTTnet.Client;
using MQTTnet.Extensions.WebSocket4Net;
using MQTTnet.Formatter;
@@ -435,4 +436,54 @@ public static class Client_Connection_Samples
Console.ReadLine();
}
}
public static async Task ConnectTls_WithCaFile()
{
var mqttFactory = new MqttFactory();
X509Certificate2Collection caChain = new X509Certificate2Collection();
caChain.ImportFromPem(mosquitto_org); // from https://test.mosquitto.org/ssl/mosquitto.org.crt
using (var mqttClient = mqttFactory.CreateMqttClient())
{
var mqttClientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("test.mosquitto.org", 8883)
.WithTlsOptions(new MqttClientTlsOptionsBuilder()
.WithTrustChain(caChain)
.Build())
.Build();
var connAck = await mqttClient.ConnectAsync(mqttClientOptions);
Console.WriteLine("Connected to test.moquitto.org:8883 with CaFile mosquitto.org.crt: " + connAck.ResultCode);
}
}
const string mosquitto_org = @"
-----BEGIN CERTIFICATE-----
MIIEAzCCAuugAwIBAgIUBY1hlCGvdj4NhBXkZ/uLUZNILAwwDQYJKoZIhvcNAQEL
BQAwgZAxCzAJBgNVBAYTAkdCMRcwFQYDVQQIDA5Vbml0ZWQgS2luZ2RvbTEOMAwG
A1UEBwwFRGVyYnkxEjAQBgNVBAoMCU1vc3F1aXR0bzELMAkGA1UECwwCQ0ExFjAU
BgNVBAMMDW1vc3F1aXR0by5vcmcxHzAdBgkqhkiG9w0BCQEWEHJvZ2VyQGF0Y2hv
by5vcmcwHhcNMjAwNjA5MTEwNjM5WhcNMzAwNjA3MTEwNjM5WjCBkDELMAkGA1UE
BhMCR0IxFzAVBgNVBAgMDlVuaXRlZCBLaW5nZG9tMQ4wDAYDVQQHDAVEZXJieTES
MBAGA1UECgwJTW9zcXVpdHRvMQswCQYDVQQLDAJDQTEWMBQGA1UEAwwNbW9zcXVp
dHRvLm9yZzEfMB0GCSqGSIb3DQEJARYQcm9nZXJAYXRjaG9vLm9yZzCCASIwDQYJ
KoZIhvcNAQEBBQADggEPADCCAQoCggEBAME0HKmIzfTOwkKLT3THHe+ObdizamPg
UZmD64Tf3zJdNeYGYn4CEXbyP6fy3tWc8S2boW6dzrH8SdFf9uo320GJA9B7U1FW
Te3xda/Lm3JFfaHjkWw7jBwcauQZjpGINHapHRlpiCZsquAthOgxW9SgDgYlGzEA
s06pkEFiMw+qDfLo/sxFKB6vQlFekMeCymjLCbNwPJyqyhFmPWwio/PDMruBTzPH
3cioBnrJWKXc3OjXdLGFJOfj7pP0j/dr2LH72eSvv3PQQFl90CZPFhrCUcRHSSxo
E6yjGOdnz7f6PveLIB574kQORwt8ePn0yidrTC1ictikED3nHYhMUOUCAwEAAaNT
MFEwHQYDVR0OBBYEFPVV6xBUFPiGKDyo5V3+Hbh4N9YSMB8GA1UdIwQYMBaAFPVV
6xBUFPiGKDyo5V3+Hbh4N9YSMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEL
BQADggEBAGa9kS21N70ThM6/Hj9D7mbVxKLBjVWe2TPsGfbl3rEDfZ+OKRZ2j6AC
6r7jb4TZO3dzF2p6dgbrlU71Y/4K0TdzIjRj3cQ3KSm41JvUQ0hZ/c04iGDg/xWf
+pp58nfPAYwuerruPNWmlStWAXf0UTqRtg4hQDWBuUFDJTuWuuBvEXudz74eh/wK
sMwfu1HFvjy5Z0iMDU8PUDepjVolOCue9ashlS4EB5IECdSR2TItnAIiIwimx839
LdUdRudafMu5T5Xma182OC0/u/xRlEm+tvKGGmfFcN0piqVl8OrSPBgIlb+1IKJE
m/XriWr/Cq4h/JfB7NTsezVslgkBaoU=
-----END CERTIFICATE-----
";
}
+3 -1
View File
@@ -184,7 +184,7 @@ public static class Client_Subscribe_Samples
* This sample shows how to achieve concurrent processing and not have message AutoAcknowledged
* This to have a proper QoS1 (at-least-once) experience for what at least MQTT specification can provide
*/
mqttClient.ApplicationMessageReceivedAsync += async ea =>
mqttClient.ApplicationMessageReceivedAsync += ea =>
{
ea.AutoAcknowledge = false;
@@ -201,6 +201,8 @@ public static class Client_Subscribe_Samples
}
_ = Task.Run(ProcessAsync, shutdownToken);
return Task.CompletedTask;
};
}
+1 -1
View File
@@ -2,7 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<IsPackable>false</IsPackable>
@@ -0,0 +1,87 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
// ReSharper disable UnusedType.Global
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
namespace MQTTnet.Samples.ManagedClient;
public sealed class Managed_Client_Subscribe_Samples
{
public static async Task Connect_Client()
{
/*
* This sample creates a simple managed MQTT client and connects to a public broker, subscribe to a topic and verifies subscription result.
*
* The managed client extends the existing _MqttClient_. It adds the following features.
* - Reconnecting when connection is lost.
* - Storing pending messages in an internal queue so that an enqueue is possible while the client remains not connected.
*/
var mqttFactory = new MqttFactory();
var subscribed = false;
using (var managedMqttClient = mqttFactory.CreateManagedMqttClient())
{
var mqttClientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("broker.hivemq.com")
.Build();
var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder()
.WithClientOptions(mqttClientOptions)
.Build();
await managedMqttClient.StartAsync(managedMqttClientOptions);
// The application message is not sent. It is stored in an internal queue and
// will be sent when the client is connected.
await managedMqttClient.EnqueueAsync("Topic", "Payload");
Console.WriteLine("The managed MQTT client is connected.");
// Wait until the queue is fully processed.
SpinWait.SpinUntil(() => managedMqttClient.PendingApplicationMessagesCount == 0, 10000);
Console.WriteLine($"Pending messages = {managedMqttClient.PendingApplicationMessagesCount}");
managedMqttClient.SubscriptionsChangedAsync += args => SubscriptionsResultAsync(args, subscribed);
await managedMqttClient.SubscribeAsync("Topic").ConfigureAwait(false);
SpinWait.SpinUntil(() => subscribed, 1000);
Console.WriteLine("Subscription properly done");
}
}
private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, bool subscribed)
{
foreach (var mqttClientSubscribeResult in arg.SubscribeResult)
{
Console.WriteLine($"Subscription reason {mqttClientSubscribeResult.ReasonString}");
foreach (var item in mqttClientSubscribeResult.Items)
{
Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}");
if (item.TopicFilter.Topic == "Topic" && item.ResultCode == MqttClientSubscribeResultCode.GrantedQoS0 && !subscribed)
{
subscribed = true;
}
}
}
foreach (var mqttClientUnsubscribeResult in arg.UnsubscribeResult)
{
Console.WriteLine($"Unsubscription reason {mqttClientUnsubscribeResult.ReasonString}");
foreach (var item in mqttClientUnsubscribeResult.Items)
{
Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}");
}
}
return Task.CompletedTask;
}
}
@@ -48,7 +48,6 @@ namespace MQTTnet.AspNetCore.Tests
{
var mockup = new ConnectionHandlerMockup();
using (var host = new WebHostBuilder().UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
.UseStartup<Startup>()
.ConfigureServices(
@@ -6,7 +6,6 @@ using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Client.Tcp;
using MQTTnet.Formatter;
using System;
using System.Net;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
@@ -22,8 +21,7 @@ namespace MQTTnet.AspNetCore.Client
{
case MqttClientTcpOptions tcpOptions:
{
var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
var tcpConnection = new TcpConnection(endpoint);
var tcpConnection = new SocketConnection(tcpOptions.RemoteEndpoint);
var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(4096, 65535));
return new MqttConnectionContext(formatter, tcpConnection);
@@ -15,15 +15,15 @@ using System.Threading.Tasks;
namespace MQTTnet.AspNetCore.Client.Tcp
{
public class TcpConnection : ConnectionContext
public sealed class SocketConnection : ConnectionContext
{
private volatile bool _aborted;
private readonly EndPoint _endPoint;
private SocketSender _sender;
private SocketReceiver _receiver;
volatile bool _aborted;
readonly EndPoint _endPoint;
SocketSender _sender;
SocketReceiver _receiver;
private Socket _socket;
private IDuplexPipe _application;
Socket _socket;
IDuplexPipe _application;
public bool IsConnected { get; private set; }
public override string ConnectionId { get; set; }
@@ -31,12 +31,12 @@ namespace MQTTnet.AspNetCore.Client.Tcp
public override IDictionary<object, object> Items { get; set; }
public override IDuplexPipe Transport { get; set; }
public TcpConnection(EndPoint endPoint)
public SocketConnection(EndPoint endPoint)
{
_endPoint = endPoint;
}
public TcpConnection(Socket socket)
public SocketConnection(Socket socket)
{
_socket = socket;
_endPoint = socket.RemoteEndPoint;
@@ -87,7 +87,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
IsConnected = true;
}
private async Task ExecuteAsync()
async Task ExecuteAsync()
{
Exception sendError = null;
try
@@ -114,7 +114,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
}
catch (Exception ex)
{
Console.WriteLine($"Unexpected exception in {nameof(TcpConnection)}.{nameof(StartAsync)}: " + ex);
Console.WriteLine($"Unexpected exception in {nameof(SocketConnection)}.{nameof(StartAsync)}: " + ex);
}
finally
{
@@ -122,7 +122,8 @@ namespace MQTTnet.AspNetCore.Client.Tcp
_application.Input.Complete(sendError);
}
}
private async Task DoReceive()
async Task DoReceive()
{
Exception error = null;
@@ -171,7 +172,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
}
}
private async Task ProcessReceives()
async Task ProcessReceives()
{
while (true)
{
@@ -204,12 +205,12 @@ namespace MQTTnet.AspNetCore.Client.Tcp
}
}
private Exception ConnectionAborted()
Exception ConnectionAborted()
{
return new MqttCommunicationException("Connection Aborted");
}
private async Task<Exception> DoSend()
async Task<Exception> DoSend()
{
Exception error = null;
@@ -240,7 +241,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
return error;
}
private async Task ProcessSends()
async Task ProcessSends()
{
while (true)
{
@@ -11,6 +11,7 @@
<Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.AspNetCore</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -37,12 +38,10 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
</ItemGroup>
@@ -51,11 +50,11 @@
<DefineConstants>RELEASE;NETSTANDARD2_0</DefineConstants>
</PropertyGroup>
<ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp3.1' or '$(TargetFramework)' == 'net5.0' or '$(TargetFramework)' == 'net6.0' or '$(TargetFramework)' == 'net7.0' ">
<ItemGroup Condition="'$(TargetFramework)' != 'netstandard2.0' and '$(TargetFramework)' != 'netcoreapp2.1' ">
<FrameworkReference Include="Microsoft.AspNetCore.App" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)' != 'netcoreapp3.1' and '$(TargetFramework)' != 'net5.0' and '$(TargetFramework)' != 'net6.0' and '$(TargetFramework)' != 'net7.0' ">
<ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0' or '$(TargetFramework)' == 'netcoreapp2.1' ">
<PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.1.0" />
</ItemGroup>
@@ -33,7 +33,7 @@ namespace MQTTnet.AspNetCore
PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
_connection = connection ?? throw new ArgumentNullException(nameof(connection));
if (!(_connection is TcpConnection tcp) || tcp.IsConnected)
if (!(_connection is SocketConnection tcp) || tcp.IsConnected)
{
_input = connection.Transport.Input;
_output = connection.Transport.Output;
@@ -83,8 +83,6 @@ namespace MQTTnet.AspNetCore
}
}
public bool IsReadingPacket { get; private set; }
public bool IsSecureConnection
{
get
@@ -111,7 +109,7 @@ namespace MQTTnet.AspNetCore
public async Task ConnectAsync(CancellationToken cancellationToken)
{
if (_connection is TcpConnection tcp && !tcp.IsConnected)
if (_connection is SocketConnection tcp && !tcp.IsConnected)
{
await tcp.StartAsync().ConfigureAwait(false);
}
@@ -164,11 +162,6 @@ namespace MQTTnet.AspNetCore
BytesReceived += received;
return packet;
}
else
{
// we did receive something but the message is not yet complete
IsReadingPacket = true;
}
}
else if (readResult.IsCompleted)
{
@@ -189,12 +182,9 @@ namespace MQTTnet.AspNetCore
// completing the channel makes sure that there is no more data read after a protocol error
_input?.Complete(exception);
_output?.Complete(exception);
throw;
}
finally
{
IsReadingPacket = false;
}
cancellationToken.ThrowIfCancellationRequested();
return null;
+14 -6
View File
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
@@ -14,20 +15,27 @@ namespace MQTTnet.AspNetCore
{
public sealed class MqttHostedServer : MqttServer, IHostedService
{
public MqttHostedServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
: base(options, adapters, logger)
readonly MqttFactory _mqttFactory;
public MqttHostedServer(MqttFactory mqttFactory, MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
options,
adapters,
logger)
{
_mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
}
public Task StartAsync(CancellationToken cancellationToken)
public async Task StartAsync(CancellationToken cancellationToken)
{
_ = StartAsync();
return Task.CompletedTask;
// The yield makes sure that the hosted service is considered up and running.
await Task.Yield();
_ = StartAsync();
}
public Task StopAsync(CancellationToken cancellationToken)
{
return StopAsync();
return StopAsync(_mqttFactory.CreateMqttServerStopOptionsBuilder().Build());
}
}
}
@@ -4,6 +4,7 @@
using System;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
@@ -14,86 +15,74 @@ namespace MQTTnet.AspNetCore
{
public static class ServiceCollectionExtensions
{
public static IServiceCollection AddMqttServer(this IServiceCollection serviceCollection, Action<MqttServerOptionsBuilder> configure = null)
{
if (serviceCollection is null)
{
throw new ArgumentNullException(nameof(serviceCollection));
}
serviceCollection.AddMqttConnectionHandler();
serviceCollection.AddHostedMqttServer(configure);
return serviceCollection;
}
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
services.AddSingleton(options);
services.AddHostedMqttServer();
return services;
}
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure = null)
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure)
{
services.AddSingleton(s =>
if (services == null)
{
var serverOptionsBuilder = new MqttServerOptionsBuilder();
configure?.Invoke(serverOptionsBuilder);
return serverOptionsBuilder.Build();
});
throw new ArgumentNullException(nameof(services));
}
services.AddHostedMqttServer();
if (configure == null)
{
throw new ArgumentNullException(nameof(configure));
}
return services;
var serverOptionsBuilder = new MqttServerOptionsBuilder();
configure.Invoke(serverOptionsBuilder);
var options = serverOptionsBuilder.Build();
return AddHostedMqttServer(services, options);
}
public static void AddHostedMqttServer(this IServiceCollection services)
{
// The user may have these services already registered.
services.TryAddSingleton<IMqttNetLogger>(MqttNetNullLogger.Instance);
services.TryAddSingleton(new MqttFactory());
services.AddSingleton<MqttHostedServer>();
services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
services.AddSingleton<MqttServer>(s => s.GetService<MqttHostedServer>());
}
public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)
{
services.AddSingleton(s =>
if (services == null)
{
var builder = new AspNetMqttServerOptionsBuilder(s);
configure(builder);
return builder.Build();
});
throw new ArgumentNullException(nameof(services));
}
services.AddSingleton(
s =>
{
var builder = new AspNetMqttServerOptionsBuilder(s);
configure(builder);
return builder.Build();
});
services.AddHostedMqttServer();
return services;
}
static IServiceCollection AddHostedMqttServer(this IServiceCollection services)
{
var logger = new MqttNetEventLogger();
services.AddSingleton<IMqttNetLogger>(logger);
services.AddSingleton<MqttHostedServer>();
services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
services.AddSingleton<MqttServer>(s => s.GetService<MqttHostedServer>());
return services;
}
public static IServiceCollection AddMqttWebSocketServerAdapter(this IServiceCollection services)
{
services.AddSingleton<MqttWebSocketServerAdapter>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttWebSocketServerAdapter>());
return services;
}
public static IServiceCollection AddMqttTcpServerAdapter(this IServiceCollection services)
{
services.AddSingleton<MqttTcpServerAdapter>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
return services;
}
public static IServiceCollection AddMqttConnectionHandler(this IServiceCollection services)
{
services.AddSingleton<MqttConnectionHandler>();
@@ -101,5 +90,44 @@ namespace MQTTnet.AspNetCore
return services;
}
public static void AddMqttLogger(this IServiceCollection services, IMqttNetLogger logger)
{
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
services.AddSingleton(logger);
}
public static IServiceCollection AddMqttServer(this IServiceCollection serviceCollection, Action<MqttServerOptionsBuilder> configure = null)
{
if (serviceCollection is null)
{
throw new ArgumentNullException(nameof(serviceCollection));
}
serviceCollection.AddMqttConnectionHandler();
serviceCollection.AddHostedMqttServer(configure);
return serviceCollection;
}
public static IServiceCollection AddMqttTcpServerAdapter(this IServiceCollection services)
{
services.AddSingleton<MqttTcpServerAdapter>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
return services;
}
public static IServiceCollection AddMqttWebSocketServerAdapter(this IServiceCollection services)
{
services.AddSingleton<MqttWebSocketServerAdapter>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttWebSocketServerAdapter>());
return services;
}
}
}
}
@@ -29,12 +29,12 @@ namespace MQTTnet.Benchmarks
var task = Task.Run(() => server.AcceptSocket());
var clientConnection = new TcpConnection(new IPEndPoint(IPAddress.Loopback, 1883));
var clientConnection = new SocketConnection(new IPEndPoint(IPAddress.Loopback, 1883));
clientConnection.StartAsync().GetAwaiter().GetResult();
_client = clientConnection.Transport;
var serverConnection = new TcpConnection(task.GetAwaiter().GetResult());
var serverConnection = new SocketConnection(task.GetAwaiter().GetResult());
serverConnection.StartAsync().GetAwaiter().GetResult();
_server = serverConnection.Transport;
}
@@ -25,6 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient
event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync;
event Func<SubscriptionsChangedEventArgs, Task> SubscriptionsChangedAsync;
IMqttClient InternalClient { get; }
bool IsConnected { get; }
@@ -0,0 +1,20 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
namespace MQTTnet.Extensions.ManagedClient
{
public sealed class InterceptingPublishMessageEventArgs : EventArgs
{
public InterceptingPublishMessageEventArgs(ManagedMqttApplicationMessage applicationMessage)
{
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
}
public ManagedMqttApplicationMessage ApplicationMessage { get; }
public bool AcceptPublish { get; set; } = true;
}
}
@@ -13,6 +13,7 @@
<Description>This is an extension library which provides a managed MQTT client with additional features using MQTTnet.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.ManagedClient</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -50,10 +51,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -19,12 +19,14 @@ namespace MQTTnet.Extensions.ManagedClient
{
public sealed class ManagedMqttClient : Disposable, IManagedMqttClient
{
readonly AsyncEvent<InterceptingPublishMessageEventArgs> _interceptingPublishMessageEvent = new AsyncEvent<InterceptingPublishMessageEventArgs>();
readonly AsyncEvent<ApplicationMessageProcessedEventArgs> _applicationMessageProcessedEvent = new AsyncEvent<ApplicationMessageProcessedEventArgs>();
readonly AsyncEvent<ApplicationMessageSkippedEventArgs> _applicationMessageSkippedEvent = new AsyncEvent<ApplicationMessageSkippedEventArgs>();
readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
readonly AsyncEvent<EventArgs> _connectionStateChangedEvent = new AsyncEvent<EventArgs>();
readonly AsyncEvent<ManagedProcessFailedEventArgs> _synchronizingSubscriptionsFailedEvent = new AsyncEvent<ManagedProcessFailedEventArgs>();
readonly AsyncEvent<SubscriptionsChangedEventArgs> _subscriptionsChangedEvent = new AsyncEvent<SubscriptionsChangedEventArgs>();
readonly MqttNetSourceLogger _logger;
readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
@@ -88,6 +90,14 @@ namespace MQTTnet.Extensions.ManagedClient
remove => _applicationMessageProcessedEvent.RemoveHandler(value);
}
public event Func<InterceptingPublishMessageEventArgs, Task> InterceptPublishMessageAsync
{
add => _interceptingPublishMessageEvent.AddHandler(value);
remove => _interceptingPublishMessageEvent.RemoveHandler(value);
}
public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
{
add => InternalClient.ApplicationMessageReceivedAsync += value;
@@ -124,6 +134,12 @@ namespace MQTTnet.Extensions.ManagedClient
remove => _synchronizingSubscriptionsFailedEvent.RemoveHandler(value);
}
public event Func<SubscriptionsChangedEventArgs, Task> SubscriptionsChangedAsync
{
add => _subscriptionsChangedEvent.AddHandler(value);
remove => _subscriptionsChangedEvent.RemoveHandler(value);
}
public IMqttClient InternalClient { get; }
public bool IsConnected => InternalClient.IsConnected;
@@ -363,6 +379,14 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
async Task HandleSubscriptionsResultAsync(SendSubscribeUnsubscribeResult subscribeUnsubscribeResult)
{
if (_subscriptionsChangedEvent.HasHandlers)
{
await _subscriptionsChangedEvent.InvokeAsync(new SubscriptionsChangedEventArgs(subscribeUnsubscribeResult.SubscribeResults, subscribeUnsubscribeResult.UnsubscribeResults)).ConfigureAwait(false);
}
}
async Task MaintainConnectionAsync(CancellationToken cancellationToken)
{
try
@@ -464,19 +488,22 @@ namespace MQTTnet.Extensions.ManagedClient
if (_reconnectSubscriptions.Any())
{
topicFilters = new List<MqttTopicFilter>();
SendSubscribeUnsubscribeResult subscribeUnsubscribeResult;
foreach (var sub in _reconnectSubscriptions)
{
topicFilters.Add(sub.Value);
if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
{
await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
topicFilters.Clear();
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
catch (Exception exception)
@@ -492,6 +519,7 @@ namespace MQTTnet.Extensions.ManagedClient
while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
{
List<MqttTopicFilter> subscriptions;
SendSubscribeUnsubscribeResult subscribeUnsubscribeResult;
HashSet<string> unsubscriptions;
lock (_subscriptions)
@@ -527,12 +555,14 @@ namespace MQTTnet.Extensions.ManagedClient
if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
{
await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
addedTopicFilters.Clear();
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
var removedTopicFilters = new List<string>();
foreach (var unSub in unsubscriptions)
@@ -541,12 +571,14 @@ namespace MQTTnet.Extensions.ManagedClient
if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
{
await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
removedTopicFilters.Clear();
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
@@ -579,8 +611,10 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
async Task SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
async Task<SendSubscribeUnsubscribeResult> SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
{
var subscribeResults = new List<MqttClientSubscribeResult>();
var unsubscribeResults = new List<MqttClientUnsubscribeResult>();
try
{
if (removedSubscriptions != null && removedSubscriptions.Any())
@@ -592,7 +626,8 @@ namespace MQTTnet.Extensions.ManagedClient
unsubscribeOptionsBuilder.WithTopicFilter(removedSubscription);
}
await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
var unsubscribeResult = await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
unsubscribeResults.Add(unsubscribeResult);
//clear because these worked, maybe the subscribe below will fail, only report those
removedSubscriptions.Clear();
@@ -607,13 +642,16 @@ namespace MQTTnet.Extensions.ManagedClient
subscribeOptionsBuilder.WithTopicFilter(addedSubscription);
}
await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
var subscribeResult = await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
subscribeResults.Add(subscribeResult);
}
}
catch (Exception exception)
{
await HandleSubscriptionExceptionAsync(exception, addedSubscriptions, removedSubscriptions).ConfigureAwait(false);
}
return new SendSubscribeUnsubscribeResult(subscribeResults, unsubscribeResults);
}
void StartPublishing()
@@ -700,9 +738,20 @@ namespace MQTTnet.Extensions.ManagedClient
async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
{
Exception transmitException = null;
bool acceptPublish = true;
try
{
await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
if (_interceptingPublishMessageEvent.HasHandlers)
{
var interceptEventArgs = new InterceptingPublishMessageEventArgs(message);
await _interceptingPublishMessageEvent.InvokeAsync(interceptEventArgs).ConfigureAwait(false);
acceptPublish = interceptEventArgs.AcceptPublish;
}
if (acceptPublish)
{
await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
}
using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync
{
@@ -0,0 +1,23 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic;
using MQTTnet.Client;
namespace MQTTnet.Extensions.ManagedClient
{
public sealed class SendSubscribeUnsubscribeResult
{
public SendSubscribeUnsubscribeResult(List<MqttClientSubscribeResult> subscribeResults, List<MqttClientUnsubscribeResult> unsubscribeResults)
{
SubscribeResults = subscribeResults ?? throw new ArgumentNullException(nameof(subscribeResults));
UnsubscribeResults = unsubscribeResults ?? throw new ArgumentNullException(nameof(unsubscribeResults));
}
public List<MqttClientSubscribeResult> SubscribeResults { get; private set; }
public List<MqttClientUnsubscribeResult> UnsubscribeResults { get; private set; }
}
}
@@ -0,0 +1,23 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Client;
using System;
using System.Collections.Generic;
namespace MQTTnet.Extensions.ManagedClient
{
public sealed class SubscriptionsChangedEventArgs : EventArgs
{
public SubscriptionsChangedEventArgs(List<MqttClientSubscribeResult> subscribeResult, List<MqttClientUnsubscribeResult> unsubscribeResult)
{
SubscribeResult = subscribeResult ?? throw new ArgumentNullException(nameof(subscribeResult));
UnsubscribeResult = unsubscribeResult ?? throw new ArgumentNullException(nameof(unsubscribeResult));
}
public List<MqttClientSubscribeResult> SubscribeResult { get; }
public List<MqttClientUnsubscribeResult> UnsubscribeResult { get; }
}
}
@@ -13,6 +13,7 @@
<Description>This is an extension library which allows executing synchronous device calls including a response using MQTTnet.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.Rpc</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -51,10 +52,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -13,6 +13,7 @@
<Description>This is an extension library which allows using _WebSocket4Net_ as transport for MQTTnet clients.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.WebSocket4Net</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -51,10 +52,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -76,7 +76,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
List<KeyValuePair<string, string>> customHeaders = null;
if (_webSocketOptions.RequestHeaders != null)
{
customHeaders = _webSocketOptions.RequestHeaders.Select(i => new KeyValuePair<string, string>(i.Key, i.Value)).ToList();
customHeaders = _webSocketOptions.RequestHeaders.ToList();
}
EndPoint proxy = null;
+2 -1
View File
@@ -4,6 +4,7 @@
using MQTTnet.Client;
using System;
using System.Net;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
@@ -27,7 +28,7 @@ namespace MQTTnet.TestApp
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "127.0.0.1"
RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
}
};
@@ -10,6 +10,7 @@
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<NoWarn>1591;NETSDK1138</NoWarn>
<LangVersion>7.3</LangVersion>
</PropertyGroup>
<ItemGroup>
+2 -1
View File
@@ -7,6 +7,7 @@ using System.Threading.Tasks;
using System.IO;
using Newtonsoft.Json;
using System.Collections.Generic;
using System.Net;
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Internal;
@@ -28,7 +29,7 @@ namespace MQTTnet.TestApp
Credentials = new RandomPassword(),
ChannelOptions = new MqttClientTcpOptions
{
Server = "broker.hivemq.com"
RemoteEndpoint = new DnsEndPoint("broker.hivemq.com", 0)
}
},
+10 -7
View File
@@ -5,13 +5,13 @@
using System;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MqttClient = MQTTnet.Client.MqttClient;
namespace MQTTnet.TestApp
{
@@ -25,7 +25,7 @@ namespace MQTTnet.TestApp
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "127.0.0.1"
RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
},
CleanSession = true
};
@@ -71,7 +71,7 @@ namespace MQTTnet.TestApp
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "127.0.0.1"
RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
}
};
@@ -114,7 +114,10 @@ namespace MQTTnet.TestApp
{
var options = new MqttClientOptions
{
ChannelOptions = new MqttClientTcpOptions { Server = "localhost" },
ChannelOptions = new MqttClientTcpOptions
{
RemoteEndpoint = new DnsEndPoint("localhost", 0)
},
ClientId = "Client1",
CleanSession = true
};
@@ -218,7 +221,7 @@ namespace MQTTnet.TestApp
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "127.0.0.1"
RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
},
CleanSession = true
};
@@ -269,7 +272,7 @@ namespace MQTTnet.TestApp
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "127.0.0.1"
RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
},
CleanSession = true
};
@@ -320,7 +323,7 @@ namespace MQTTnet.TestApp
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "127.0.0.1"
RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
},
CleanSession = true
};
+32 -23
View File
@@ -42,45 +42,45 @@ namespace MQTTnet.TestApp
};
// mqtt.eclipseprojects.io
await ExecuteTestAsync(
await ExecuteTestsAsync(
"mqtt.eclipseprojects.io TCP",
new MqttClientOptionsBuilder().WithTcpServer("mqtt.eclipseprojects.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"mqtt.eclipseprojects.io WS",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:80/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
#if NET5_0_OR_GREATER
await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13",
await ExecuteTestsAsync("mqtt.eclipseprojects.io WS TLS13",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:443/mqtt"))
.WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13 (WebSocket4Net)",
await ExecuteTestsAsync("mqtt.eclipseprojects.io WS TLS13 (WebSocket4Net)",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:443/mqtt"))
.WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build(),
true);
#endif
// test.mosquitto.org
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org TCP",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org TCP - Authenticated",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1884).WithCredentials("rw", "readwrite").WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org TCP TLS12",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
#if NET5_0_OR_GREATER
await ExecuteTestAsync("test.mosquitto.org TCP TLS13",
await ExecuteTestsAsync("test.mosquitto.org TCP TLS13",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883)
.WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
#endif
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org TCP TLS12 - Authenticated",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8885)
.WithCredentials("rw", "readwrite")
@@ -88,67 +88,67 @@ namespace MQTTnet.TestApp
.WithTlsOptions(unsafeTls12)
.Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org WS",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org WS (WebSocket4Net)",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
true);
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org WS TLS12",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"test.mosquitto.org WS TLS12 (WebSocket4Net)",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
true);
// broker.emqx.io
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.emqx.io TCP",
new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.emqx.io TCP TLS12",
new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
#if NET5_0_OR_GREATER
await ExecuteTestAsync("broker.emqx.io TCP TLS13",
await ExecuteTestsAsync("broker.emqx.io TCP TLS13",
new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883)
.WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
#endif
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.emqx.io WS",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8083/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.emqx.io WS (WebSocket4Net)",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
true);
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.emqx.io WS TLS12",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.emqx.io WS TLS12 (WebSocket4Net)",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
true);
// broker.hivemq.com
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.hivemq.com TCP",
new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.hivemq.com WS",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync(
await ExecuteTestsAsync(
"broker.hivemq.com WS (WebSocket4Net)",
new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
true);
@@ -160,6 +160,15 @@ namespace MQTTnet.TestApp
Console.ReadLine();
}
static async Task ExecuteTestsAsync(string name, MqttClientOptions options, bool useWebSocket4Net = false)
{
options.ProtocolVersion = MqttProtocolVersion.V311;
await ExecuteTestAsync(name + " V3.1.1", options, useWebSocket4Net);
options.ProtocolVersion = MqttProtocolVersion.V500;
await ExecuteTestAsync(name + " V5.0.0", options, useWebSocket4Net);
}
static async Task ExecuteTestAsync(string name, MqttClientOptions options, bool useWebSocket4Net = false)
{
try
@@ -13,6 +13,7 @@ using MQTTnet.Exceptions;
using MQTTnet.LowLevelClient;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Clients.LowLevelMqttClient
{
@@ -22,99 +22,6 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
[TestClass]
public sealed class ManagedMqttClient_Tests : BaseTestClass
{
[TestMethod]
public async Task Expose_Custom_Connection_Error()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
server.ValidatingConnectionAsync += args =>
{
args.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
return CompletedTask.Instance;
};
var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
MqttClientDisconnectedEventArgs disconnectedArgs = null;
managedClient.DisconnectedAsync += args =>
{
disconnectedArgs = args;
return CompletedTask.Instance;
};
var clientOptions = testEnvironment.Factory.CreateManagedMqttClientOptionsBuilder().WithClientOptions(testEnvironment.CreateDefaultClientOptions()).Build();
await managedClient.StartAsync(clientOptions);
await LongTestDelay();
Assert.IsNotNull(disconnectedArgs);
Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, disconnectedArgs.ConnectResult.ResultCode);
}
}
[TestMethod]
public async Task Receive_While_Not_Cleanly_Disconnected()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
await testEnvironment.StartServer(o => o.WithPersistentSessions());
var senderClient = await testEnvironment.ConnectClient();
// Prepare managed client.
var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
await managedClient.SubscribeAsync("#");
var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
var managedClientOptions = new ManagedMqttClientOptions
{
ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
.WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
.WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
.WithCleanSession(false)
.Build()
};
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
await LongTestDelay();
// Send test data.
await senderClient.PublishStringAsync("topic1");
await LongTestDelay();
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(1);
// Stop the managed client but keep session at server (not clean disconnect required).
await managedClient.StopAsync(false);
await LongTestDelay();
// Send new messages in the meantime.
await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
await LongTestDelay();
// Start the managed client, it should receive the new message.
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(2);
// Stop and start again, no new message should be received.
for (var i = 0; i < 3; i++)
{
await managedClient.StopAsync(false);
await LongTestDelay();
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
}
receivedMessages.AssertReceivedCountEquals(2);
}
}
[TestMethod]
public async Task Connect_To_Invalid_Server()
{
@@ -181,6 +88,38 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
}
}
[TestMethod]
public async Task Expose_Custom_Connection_Error()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
server.ValidatingConnectionAsync += args =>
{
args.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
return CompletedTask.Instance;
};
var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
MqttClientDisconnectedEventArgs disconnectedArgs = null;
managedClient.DisconnectedAsync += args =>
{
disconnectedArgs = args;
return CompletedTask.Instance;
};
var clientOptions = testEnvironment.Factory.CreateManagedMqttClientOptionsBuilder().WithClientOptions(testEnvironment.CreateDefaultClientOptions()).Build();
await managedClient.StartAsync(clientOptions);
await LongTestDelay();
Assert.IsNotNull(disconnectedArgs);
Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, disconnectedArgs.ConnectResult.ResultCode);
}
}
[TestMethod]
public async Task ManagedClients_Will_Message_Send()
{
@@ -224,6 +163,67 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
}
}
[TestMethod]
public async Task Receive_While_Not_Cleanly_Disconnected()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
await testEnvironment.StartServer(o => o.WithPersistentSessions());
var senderClient = await testEnvironment.ConnectClient();
// Prepare managed client.
var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
await managedClient.SubscribeAsync("#");
var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
var managedClientOptions = new ManagedMqttClientOptions
{
ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
.WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
.WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
.WithCleanSession(false)
.Build()
};
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
await LongTestDelay();
// Send test data.
await senderClient.PublishStringAsync("topic1");
await LongTestDelay();
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(1);
// Stop the managed client but keep session at server (not clean disconnect required).
await managedClient.StopAsync(false);
await LongTestDelay();
// Send new messages in the meantime.
await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
await LongTestDelay();
// Start the managed client, it should receive the new message.
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(2);
// Stop and start again, no new message should be received.
for (var i = 0; i < 3; i++)
{
await managedClient.StopAsync(false);
await LongTestDelay();
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
}
receivedMessages.AssertReceivedCountEquals(2);
}
}
[TestMethod]
public async Task Start_Stop()
{
@@ -375,7 +375,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort);
var receivedManagedMessages = new List<MqttApplicationMessage>();
var managedClient = testEnvironment.Factory.CreateManagedMqttClient(testEnvironment.CreateClient());
managedClient.ApplicationMessageReceivedAsync += e =>
{
@@ -403,7 +403,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
// Make sure that it gets received after subscribing again.
await managedClient.SubscribeAsync("topic");
await LongTestDelay();
Assert.AreEqual(2, receivedManagedMessages.Count);
}
}
@@ -421,7 +421,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
var receivingClient = await CreateManagedClientAsync(testEnvironment, null, connectionCheckInterval);
var sendingClient = await testEnvironment.ConnectClient();
await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", PayloadSegment = new ArraySegment<byte>( new byte[] { 1 }), Retain = true });
await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", PayloadSegment = new ArraySegment<byte>(new byte[] { 1 }), Retain = true });
var subscribeTime = DateTime.UtcNow;
@@ -437,6 +437,109 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
}
}
[TestMethod]
public async Task ManagedClients_CanInterceptPublishedMessage_PreventingPublish()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var receivedMessagesCount = 0;
var interceptedMessagesCount = 0;
var clientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
.Build();
var sendingClient = testEnvironment.CreateClient();
var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
// Wait until the managed client is fully set up and running.
await Task.Delay(1000);
var receivingClient = await testEnvironment.ConnectClient();
receivingClient.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return CompletedTask.Instance;
};
sendingManagedClient.InterceptPublishMessageAsync += e =>
{
Interlocked.Increment(ref interceptedMessagesCount);
e.AcceptPublish = false;
return CompletedTask.Instance;
};
await receivingClient.SubscribeAsync("My/last/will");
// Wait for arrival of the will message at the receiver.
await Task.Delay(5000);
await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
// Wait for arrival of the will message at the receiver.
await Task.Delay(5000);
Assert.AreEqual(0, receivedMessagesCount);
Assert.AreEqual(1, interceptedMessagesCount);
}
}
[TestMethod]
public async Task ManagedClients_CanInterceptPublishedMessage_AllowingPublish()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var receivedMessagesCount = 0;
var interceptedMessagesCount = 0;
var clientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
.Build();
var sendingClient = testEnvironment.CreateClient();
var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
// Wait until the managed client is fully set up and running.
await Task.Delay(1000);
var receivingClient = await testEnvironment.ConnectClient();
receivingClient.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return CompletedTask.Instance;
};
sendingManagedClient.InterceptPublishMessageAsync += e =>
{
Interlocked.Increment(ref interceptedMessagesCount);
e.AcceptPublish = true;
return CompletedTask.Instance;
};
await receivingClient.SubscribeAsync("My/last/will");
// Wait for arrival of the will message at the receiver.
await Task.Delay(5000);
await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
// Wait for arrival of the will message at the receiver.
await Task.Delay(5000);
Assert.AreEqual(1, receivedMessagesCount);
Assert.AreEqual(1, interceptedMessagesCount);
}
}
// This case also serves as a regression test for the previous behavior which re-published
// each and every existing subscriptions with every new subscription that was made
// (causing performance problems and having the visible symptom of retained messages being received again)
@@ -65,7 +65,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
return CompletedTask.Instance;
};
await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build());
await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").Build());
Assert.Fail("Must fail!");
}
@@ -167,6 +167,38 @@ namespace MQTTnet.Tests.Clients.MqttClient
}
}
[TestMethod]
public async Task No_Unobserved_Exception()
{
using (var testEnvironment = CreateTestEnvironment())
{
testEnvironment.IgnoreClientLogErrors = true;
var client = testEnvironment.CreateClient();
var options = new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").WithTimeout(TimeSpan.FromSeconds(2)).Build();
try
{
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(0.5)))
{
await client.ConnectAsync(options, timeout.Token);
}
}
catch (OperationCanceledException)
{
}
client.Dispose();
// These delays and GC calls are required in order to make calling the finalizer reproducible.
GC.Collect();
GC.WaitForPendingFinalizers();
await LongTestDelay();
await LongTestDelay();
await LongTestDelay();
}
}
[TestMethod]
public async Task Return_Non_Success()
{
@@ -208,9 +240,9 @@ namespace MQTTnet.Tests.Clients.MqttClient
await mqttClient.SubscribeAsync("test", MqttQualityOfServiceLevel.AtLeastOnce);
}
}
catch (MqttCommunicationException exception)
catch (MqttClientNotConnectedException exception)
{
if (exception.Message == "The client is not connected.")
if (exception.Message == "The MQTT client is not connected.")
{
return;
}
@@ -17,6 +17,7 @@ using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
// ReSharper disable InconsistentNaming
+2 -2
View File
@@ -52,7 +52,7 @@ namespace MQTTnet.Tests.Extensions
var paramValue = "123";
var parameters = new Dictionary<string, object>
{
{ TestParametersTopicGenerationStrategy.ExpectedParamName, "123" },
{ TestParametersTopicGenerationStrategy.ExpectedParamName, "123" }
};
using (var testEnvironment = CreateTestEnvironment())
@@ -164,7 +164,7 @@ namespace MQTTnet.Tests.Extensions
using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
{
var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
}
}
}
+2 -1
View File
@@ -1,7 +1,8 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net452;net461;net48;netcoreapp3.1;net6.0;net7.0</TargetFrameworks>
<TargetFrameworks>netcoreapp3.1;net7.0</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net48</TargetFrameworks>
<IsPackable>false</IsPackable>
<LangVersion>7.3</LangVersion>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
+1 -1
View File
@@ -273,7 +273,7 @@ namespace MQTTnet.Tests.MQTTv5
Assert.AreEqual(applicationMessage.ResponseTopic, receivedMessage.ResponseTopic);
Assert.AreEqual(applicationMessage.MessageExpiryInterval, receivedMessage.MessageExpiryInterval);
CollectionAssert.AreEqual(applicationMessage.CorrelationData, receivedMessage.CorrelationData);
CollectionAssert.AreEqual(applicationMessage.Payload, receivedMessage.Payload);
CollectionAssert.AreEqual(applicationMessage.PayloadSegment.ToArray(), receivedMessage.PayloadSegment.ToArray());
CollectionAssert.AreEqual(applicationMessage.UserProperties, receivedMessage.UserProperties);
}
}
@@ -10,6 +10,7 @@ using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.MQTTv5
{
+87 -39
View File
@@ -3,7 +3,6 @@
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
@@ -25,9 +24,9 @@ namespace MQTTnet.Tests.Mockups
public sealed class TestEnvironment : IDisposable
{
readonly List<string> _clientErrors = new List<string>();
readonly ConcurrentBag<IMqttClient> _clients = new ConcurrentBag<IMqttClient>();
readonly List<IMqttClient> _clients = new List<IMqttClient>();
readonly List<Exception> _exceptions = new List<Exception>();
readonly ConcurrentBag<ILowLevelMqttClient> _lowLevelClients = new ConcurrentBag<ILowLevelMqttClient>();
readonly List<ILowLevelMqttClient> _lowLevelClients = new List<ILowLevelMqttClient>();
readonly MqttProtocolVersion _protocolVersion;
readonly List<string> _serverErrors = new List<string>();
@@ -40,6 +39,8 @@ namespace MQTTnet.Tests.Mockups
_protocolVersion = protocolVersion;
TestContext = testContext;
TaskScheduler.UnobservedTaskException += TrackUnobservedTaskException;
ServerLogger.LogMessagePublished += (s, e) =>
{
if (Debugger.IsAttached)
@@ -216,8 +217,9 @@ namespace MQTTnet.Tests.Mockups
public IMqttClient CreateClient()
{
var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : MqttNetNullLogger.Instance;
var client = Factory.CreateMqttClient(logger);
client.ConnectingAsync += e =>
{
if (TestContext != null)
@@ -232,9 +234,12 @@ namespace MQTTnet.Tests.Mockups
return CompletedTask.Instance;
};
_clients.Add(client);
lock (_clients)
{
_clients.Add(client);
}
return client;
}
@@ -245,13 +250,20 @@ namespace MQTTnet.Tests.Mockups
public MqttClientOptionsBuilder CreateDefaultClientOptionsBuilder()
{
return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort).WithClientId(TestContext.TestName + "_" + Guid.NewGuid());
return Factory.CreateClientOptionsBuilder()
.WithProtocolVersion(_protocolVersion)
.WithTcpServer("127.0.0.1", ServerPort)
.WithClientId(TestContext.TestName + "_" + Guid.NewGuid());
}
public ILowLevelMqttClient CreateLowLevelClient()
{
var client = Factory.CreateLowLevelMqttClient(ClientLogger);
_lowLevelClients.Add(client);
lock (_lowLevelClients)
{
_lowLevelClients.Add(client);
}
return client;
}
@@ -287,45 +299,68 @@ namespace MQTTnet.Tests.Mockups
public void Dispose()
{
foreach (var mqttClient in _clients)
try
{
lock (_clients)
{
foreach (var mqttClient in _clients)
{
try
{
//mqttClient.DisconnectAsync().GetAwaiter().GetResult();
}
catch
{
// This can happen when the test already disconnected the client.
}
finally
{
mqttClient?.Dispose();
}
}
_clients.Clear();
}
lock (_lowLevelClients)
{
foreach (var lowLevelMqttClient in _lowLevelClients)
{
lowLevelMqttClient.Dispose();
}
_lowLevelClients.Clear();
}
try
{
//mqttClient.DisconnectAsync().GetAwaiter().GetResult();
Server?.StopAsync().GetAwaiter().GetResult();
}
catch
{
// This can happen when the test already disconnected the client.
// This can happen when the test already stopped the server.
}
finally
{
mqttClient?.Dispose();
Server?.Dispose();
}
}
foreach (var lowLevelMqttClient in _lowLevelClients)
{
lowLevelMqttClient.Dispose();
}
Server = null;
try
{
Server?.StopAsync().GetAwaiter().GetResult();
}
catch
{
// This can happen when the test already stopped the server.
ThrowIfLogErrors();
GC.Collect();
GC.WaitForFullGCComplete();
GC.WaitForPendingFinalizers();
if (_exceptions.Any())
{
throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
}
}
finally
{
Server?.Dispose();
}
ThrowIfLogErrors();
if (_exceptions.Any())
{
throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
TaskScheduler.UnobservedTaskException -= TrackUnobservedTaskException;
}
}
@@ -342,7 +377,7 @@ namespace MQTTnet.Tests.Mockups
var options = optionsBuilder.Build();
var server = CreateServer(options);
await server.StartAsync();
await server.StartAsync().ConfigureAwait(false);
// The OS has chosen the port to we have to properly expose it to the tests.
ServerPort = options.DefaultEndpointOptions.Port;
@@ -370,13 +405,16 @@ namespace MQTTnet.Tests.Mockups
public void ThrowIfLogErrors()
{
lock (_serverErrors)
if (!IgnoreServerLogErrors)
{
if (!IgnoreServerLogErrors && _serverErrors.Count > 0)
lock (_serverErrors)
{
var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)}).";
Console.WriteLine(message);
throw new Exception(message);
if (_serverErrors.Count > 0)
{
var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)}).";
Console.WriteLine(message);
throw new Exception(message);
}
}
}
@@ -396,10 +434,20 @@ namespace MQTTnet.Tests.Mockups
public void TrackException(Exception exception)
{
if (exception == null)
{
return;
}
lock (_exceptions)
{
_exceptions.Add(exception);
}
}
void TrackUnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs e)
{
TrackException(e.Exception);
}
}
}
@@ -1,3 +1,5 @@
#pragma warning disable CS0618 // Type or member is obsolete
using Microsoft.VisualStudio.TestTools.UnitTesting;
using System;
using System.Collections.Generic;
@@ -8,7 +10,7 @@ using System.Threading.Tasks;
namespace MQTTnet.Tests
{
[TestClass]
public class MqttApplicationMessageTest
public sealed class MqttApplicationMessageTest
{
[TestMethod]
public void PayloadSegment()
@@ -17,32 +19,27 @@ namespace MQTTnet.Tests
Assert.AreEqual(0, message.Payload.Length);
Assert.AreEqual(0, message.PayloadSegment.Count);
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
message.Payload = new byte[] { 1, 2 };
Assert.AreEqual(2, message.Payload.Length);
Assert.AreEqual(2, message.PayloadSegment.Count);
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
message.Payload = new byte[] { 1, 2, 3 };
Assert.AreEqual(3, message.Payload.Length);
Assert.AreEqual(3, message.PayloadSegment.Count);
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
message.PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3 });
Assert.AreEqual(3, message.Payload.Length);
Assert.AreEqual(3, message.PayloadSegment.Count);
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
message.PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3 }, 1, 1);
Assert.AreEqual(1, message.Payload.Length);
Assert.AreEqual(1, message.PayloadSegment.Count);
Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
Assert.IsFalse(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
Assert.IsFalse(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
}
}
}
+4 -4
View File
@@ -303,7 +303,7 @@ namespace MQTTnet.Tests.Server
var server = await testEnvironment.StartServer();
server.InterceptingPublishAsync += e =>
{
e.ApplicationMessage.Payload = Encoding.ASCII.GetBytes("extended");
e.ApplicationMessage.PayloadSegment = new ArraySegment<byte>(Encoding.ASCII.GetBytes("extended"));
return CompletedTask.Instance;
};
@@ -314,7 +314,7 @@ namespace MQTTnet.Tests.Server
var isIntercepted = false;
c2.ApplicationMessageReceivedAsync += e =>
{
isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.Payload), StringComparison.Ordinal) == 0;
isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment.ToArray()), StringComparison.Ordinal) == 0;
return CompletedTask.Instance;
};
@@ -425,7 +425,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage
{
Topic = "/test/1",
Payload = Encoding.UTF8.GetBytes("true"),
PayloadSegment = new ArraySegment<byte>(Encoding.UTF8.GetBytes("true")),
QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
})
{
@@ -780,7 +780,7 @@ namespace MQTTnet.Tests.Server
var client1 = await testEnvironment.ConnectClient();
client1.ApplicationMessageReceivedAsync += e =>
{
receivedBody = e.ApplicationMessage.Payload;
receivedBody = e.ApplicationMessage.PayloadSegment.ToArray();
return CompletedTask.Instance;
};
@@ -1,4 +1,4 @@
#if !(NET452 || NET461)
#if !(NET452 || NET461 || NET48)
using System;
using System.Collections.Concurrent;
using System.Diagnostics;
@@ -149,18 +149,14 @@ namespace MQTTnet.Tests.Server
}
}
class ClientTestHarness : IDisposable
sealed class ClientTestHarness : IDisposable
{
readonly HotSwappableClientCertProvider _hotSwapClient = new HotSwappableClientCertProvider();
IMqttClient _client;
public string ClientId => _client.Options.ClientId;
public void ClearServerCerts()
{
_hotSwapClient.ClearServerCerts();
}
public Task Connect()
{
return Run_Client_Connection();
@@ -169,6 +165,7 @@ namespace MQTTnet.Tests.Server
public void Dispose()
{
_client.Dispose();
_hotSwapClient.Dispose();
}
public X509Certificate2 GetCurrentClientCert()
@@ -189,6 +186,8 @@ namespace MQTTnet.Tests.Server
public void WaitForConnectOrFail(TimeSpan timeout)
{
Thread.Sleep(100);
if (!_client.IsConnected)
{
_client.ReconnectAsync().Wait(timeout);
@@ -210,12 +209,12 @@ namespace MQTTnet.Tests.Server
Assert.IsFalse(_client.IsConnected, "Client connection success but test wanted fail");
}
public void WaitForDisconnect(TimeSpan timeout)
void WaitForDisconnect(TimeSpan timeout)
{
var timer = Stopwatch.StartNew();
while ((_client == null || _client.IsConnected) && timer.Elapsed < timeout)
{
Thread.Sleep(5);
Thread.Sleep(100);
}
}
@@ -252,22 +251,17 @@ namespace MQTTnet.Tests.Server
var timer = Stopwatch.StartNew();
while ((_client == null || !_client.IsConnected) && timer.Elapsed < timeout)
{
Thread.Sleep(5);
Thread.Sleep(100);
}
}
}
class ServerTestHarness : IDisposable
sealed class ServerTestHarness : IDisposable
{
readonly HotSwappableServerCertProvider _hotSwapServer = new HotSwappableServerCertProvider();
MqttServer _server;
public void ClearClientCerts()
{
_hotSwapServer.ClearClientCerts();
}
public void Dispose()
{
if (_server != null)
@@ -276,15 +270,12 @@ namespace MQTTnet.Tests.Server
_server.Dispose();
}
if (_hotSwapServer != null)
{
_hotSwapServer.Dispose();
}
_hotSwapServer?.Dispose();
}
public async Task ForceDisconnectAsync(ClientTestHarness client)
public Task ForceDisconnectAsync(ClientTestHarness client)
{
await _server.DisconnectClientAsync(client.ClientId, MqttDisconnectReasonCode.UnspecifiedError);
return _server.DisconnectClientAsync(client.ClientId, MqttDisconnectReasonCode.UnspecifiedError);
}
public X509Certificate2 GetCurrentServerCert()
@@ -302,7 +293,7 @@ namespace MQTTnet.Tests.Server
_hotSwapServer.InstallNewClientCert(serverCert);
}
public async Task StartServer()
public Task StartServer()
{
var mqttFactory = new MqttFactory();
@@ -312,24 +303,33 @@ namespace MQTTnet.Tests.Server
.Build();
mqttServerOptions.TlsEndpointOptions.ClientCertificateRequired = true;
_server = mqttFactory.CreateMqttServer(mqttServerOptions);
await _server.StartAsync();
return _server.StartAsync();
}
}
class HotSwappableClientCertProvider : IMqttClientCertificatesProvider
class HotSwappableClientCertProvider : IMqttClientCertificatesProvider, IDisposable
{
X509Certificate2Collection _certificates;
ConcurrentBag<X509Certificate2> ServerCerts = new ConcurrentBag<X509Certificate2>();
ConcurrentBag<X509Certificate2> _serverCerts = new ConcurrentBag<X509Certificate2>();
public HotSwappableClientCertProvider()
{
_certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
}
public void ClearServerCerts()
public void Dispose()
{
ServerCerts = new ConcurrentBag<X509Certificate2>();
if (_certificates != null)
{
foreach (var certs in _certificates)
{
#if !NET452
certs.Dispose();
#endif
}
}
}
public X509CertificateCollection GetCertificates()
@@ -339,18 +339,17 @@ namespace MQTTnet.Tests.Server
public void HotSwapCert()
{
var newCert = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
var oldCerts = Interlocked.Exchange(ref _certificates, newCert);
_certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
}
public void InstallNewServerCert(X509Certificate2 serverCert)
{
ServerCerts.Add(serverCert);
_serverCerts.Add(serverCert);
}
public bool OnCertificateValidation(MqttClientCertificateValidationEventArgs certContext)
{
var serverCerts = ServerCerts.ToArray();
var serverCerts = _serverCerts.ToArray();
var providedCert = certContext.Certificate.GetRawCertData();
for (int i = 0, n = serverCerts.Length; i < n; i++)
@@ -365,36 +364,18 @@ namespace MQTTnet.Tests.Server
return false;
}
void Dispose()
{
if (_certificates != null)
{
foreach (var certs in _certificates)
{
#if !NET452
certs.Dispose();
#endif
}
}
}
}
class HotSwappableServerCertProvider : ICertificateProvider, IDisposable
sealed class HotSwappableServerCertProvider : ICertificateProvider, IDisposable
{
readonly ConcurrentBag<X509Certificate2> _clientCerts = new ConcurrentBag<X509Certificate2>();
X509Certificate2 _certificate;
ConcurrentBag<X509Certificate2> ClientCerts = new ConcurrentBag<X509Certificate2>();
public HotSwappableServerCertProvider()
{
_certificate = CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.1");
}
public void ClearClientCerts()
{
ClientCerts = new ConcurrentBag<X509Certificate2>();
}
public void Dispose()
{
#if !NET452
@@ -418,12 +399,12 @@ namespace MQTTnet.Tests.Server
public void InstallNewClientCert(X509Certificate2 certificate)
{
ClientCerts.Add(certificate);
_clientCerts.Add(certificate);
}
public bool RemoteCertificateValidationCallback(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
{
var serverCerts = ClientCerts.ToArray();
var serverCerts = _clientCerts.ToArray();
var providedCert = certificate.GetRawCertData();
for (int i = 0, n = serverCerts.Length; i < n; i++)
@@ -8,6 +8,7 @@ using MQTTnet.Client;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Server
{
+1 -1
View File
@@ -283,7 +283,7 @@ namespace MQTTnet.Tests.Server
server.InterceptingPublishAsync += e =>
{
e.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(e.SessionItems["default_payload"] as string ?? string.Empty);
e.ApplicationMessage.PayloadSegment = new ArraySegment<byte>(Encoding.UTF8.GetBytes(e.SessionItems["default_payload"] as string ?? string.Empty));
return CompletedTask.Instance;
};
+9 -11
View File
@@ -101,7 +101,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage
{
Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 }
PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
});
await testEnvironment.Server.InjectApplicationMessage(
@@ -109,7 +109,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage
{
Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 }
PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
}));
certificateProvider.CurrentCertificate = CreateCertificate(secondOid);
@@ -137,7 +137,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage
{
Topic = "TestTopic2",
Payload = new byte[] { 1, 2, 3, 4 }
PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
});
await testEnvironment.Server.InjectApplicationMessage(
@@ -145,7 +145,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage
{
Topic = "TestTopic2",
Payload = new byte[] { 1, 2, 3, 4 }
PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
}));
// Ensure first client still works
@@ -153,7 +153,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage
{
Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 }
PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
});
await testEnvironment.Server.InjectApplicationMessage(
@@ -161,7 +161,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage
{
Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 }
PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
}));
await Task.Delay(1000);
@@ -178,12 +178,10 @@ namespace MQTTnet.Tests.Server
var clientOptionsBuilder = testEnvironment.Factory.CreateClientOptionsBuilder();
clientOptionsBuilder.WithClientId(Guid.NewGuid().ToString())
.WithTcpServer("localhost", 8883)
.WithTls(
tls =>
.WithTlsOptions(
o =>
{
tls.UseTls = true;
tls.SslProtocol = SslProtocols.Tls12;
tls.CertificateValidationHandler = certValidator;
o.WithSslProtocols(SslProtocols.Tls12).WithCertificateValidationHandler(certValidator);
});
var clientOptions = clientOptionsBuilder.Build();
@@ -25,8 +25,6 @@ namespace MQTTnet.Adapter
long BytesReceived { get; }
bool IsReadingPacket { get; }
Task ConnectAsync(CancellationToken cancellationToken);
Task DisconnectAsync(CancellationToken cancellationToken);
@@ -14,6 +14,7 @@ namespace MQTTnet.Adapter
Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
Task StartAsync(MqttServerOptions options, IMqttNetLogger logger);
Task StopAsync();
}
}
+64 -49
View File
@@ -26,9 +26,7 @@ namespace MQTTnet.Adapter
readonly IMqttChannel _channel;
readonly byte[] _fixedHeaderBuffer = new byte[2];
readonly MqttNetSourceLogger _logger;
readonly byte[] _singleByteBuffer = new byte[1];
readonly AsyncLock _syncRoot = new AsyncLock();
Statistics _statistics; // mutable struct, don't make readonly!
@@ -47,6 +45,8 @@ namespace MQTTnet.Adapter
_logger = logger.WithSource(nameof(MqttChannelAdapter));
}
public bool AllowPacketFragmentation { get; set; } = true;
public long BytesReceived => Volatile.Read(ref _statistics._bytesReceived);
public long BytesSent => Volatile.Read(ref _statistics._bytesSent);
@@ -55,16 +55,12 @@ namespace MQTTnet.Adapter
public string Endpoint => _channel.Endpoint;
public bool IsReadingPacket { get; private set; }
public bool IsSecureConnection => _channel.IsSecureConnection;
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
public MqttPacketInspector PacketInspector { get; set; }
public bool AllowPacketFragmentation { get; set; } = true;
public async Task ConnectAsync(CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
@@ -78,12 +74,30 @@ namespace MQTTnet.Adapter
* block forever. Even a cancellation token is not supported properly.
*/
var connectTask = _channel.ConnectAsync(cancellationToken);
var timeout = new TaskCompletionSource<object>();
using (cancellationToken.Register(() => timeout.TrySetResult(null)))
{
var connectTask = Task.Run(
async () =>
{
try
{
await _channel.ConnectAsync(cancellationToken).ConfigureAwait(false);
}
catch
{
// If the timeout is already reached the exception is no longer of interest and
// must be catched. Otherwise it will arrive at the TaskScheduler.UnobservedTaskException.
if (!timeout.Task.IsCompleted)
{
throw;
}
}
},
CancellationToken.None);
await Task.WhenAny(connectTask, timeout.Task).ConfigureAwait(false);
if (timeout.Task.IsCompleted && !connectTask.IsCompleted)
{
throw new OperationCanceledException("MQTT connect canceled.", cancellationToken);
@@ -127,7 +141,8 @@ namespace MQTTnet.Adapter
try
{
PacketInspector?.BeginReceivePacket();
var localPacketInspector = PacketInspector;
localPacketInspector?.BeginReceivePacket();
ReceivedMqttPacket receivedPacket;
var receivedPacketTask = ReceiveAsync(cancellationToken);
@@ -145,7 +160,10 @@ namespace MQTTnet.Adapter
return null;
}
PacketInspector?.EndReceivePacket();
if (localPacketInspector != null)
{
await localPacketInspector.EndReceivePacket().ConfigureAwait(false);
}
Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength);
@@ -201,10 +219,15 @@ namespace MQTTnet.Adapter
try
{
var packetBuffer = PacketFormatterAdapter.Encode(packet);
PacketInspector?.BeginSendPacket(packetBuffer);
_logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
var localPacketInspector = PacketInspector;
if (localPacketInspector != null)
{
await localPacketInspector.BeginSendPacket(packetBuffer).ConfigureAwait(false);
}
_logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation)
{
await _channel.WriteAsync(packetBuffer.Join(), true, cancellationToken).ConfigureAwait(false);
@@ -373,53 +396,45 @@ namespace MQTTnet.Adapter
return ReceivedMqttPacket.Empty;
}
IsReadingPacket = true;
try
var fixedHeader = readFixedHeaderResult.FixedHeader;
if (fixedHeader.RemainingLength == 0)
{
var fixedHeader = readFixedHeaderResult.FixedHeader;
if (fixedHeader.RemainingLength == 0)
return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
}
var bodyLength = fixedHeader.RemainingLength;
var body = new byte[bodyLength];
var bodyOffset = 0;
var chunkSize = Math.Min(ReadBufferSize, bodyLength);
do
{
var bytesLeft = body.Length - bodyOffset;
if (chunkSize > bytesLeft)
{
return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
chunkSize = bytesLeft;
}
var bodyLength = fixedHeader.RemainingLength;
var body = new byte[bodyLength];
var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
var bodyOffset = 0;
var chunkSize = Math.Min(ReadBufferSize, bodyLength);
do
if (cancellationToken.IsCancellationRequested)
{
var bytesLeft = body.Length - bodyOffset;
if (chunkSize > bytesLeft)
{
chunkSize = bytesLeft;
}
return ReceivedMqttPacket.Empty;
}
var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
if (readBytes == 0)
{
return ReceivedMqttPacket.Empty;
}
if (cancellationToken.IsCancellationRequested)
{
return ReceivedMqttPacket.Empty;
}
bodyOffset += readBytes;
} while (bodyOffset < bodyLength);
if (readBytes == 0)
{
return ReceivedMqttPacket.Empty;
}
PacketInspector?.FillReceiveBuffer(body);
bodyOffset += readBytes;
} while (bodyOffset < bodyLength);
PacketInspector?.FillReceiveBuffer(body);
var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
}
finally
{
IsReadingPacket = false;
}
var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
}
static bool WrapAndThrowException(Exception exception)
+30 -25
View File
@@ -4,6 +4,7 @@
using System;
using System.IO;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Internal;
@@ -12,16 +13,20 @@ namespace MQTTnet.Adapter
{
public sealed class MqttPacketInspector
{
readonly MqttNetSourceLogger _logger;
readonly AsyncEvent<InspectMqttPacketEventArgs> _asyncEvent;
readonly MqttNetSourceLogger _logger;
MemoryStream _receivedPacketBuffer;
public MqttPacketInspector(AsyncEvent<InspectMqttPacketEventArgs> asyncEvent, IMqttNetLogger logger)
{
_asyncEvent = asyncEvent ?? throw new ArgumentNullException(nameof(asyncEvent));
if (logger == null) throw new ArgumentNullException(nameof(logger));
if (logger == null)
{
throw new ArgumentNullException(nameof(logger));
}
_logger = logger.WithSource(nameof(MqttPacketInspector));
}
@@ -36,28 +41,15 @@ namespace MQTTnet.Adapter
{
_receivedPacketBuffer = new MemoryStream();
}
_receivedPacketBuffer?.SetLength(0);
}
public void EndReceivePacket()
public Task BeginSendPacket(MqttPacketBuffer buffer)
{
if (!_asyncEvent.HasHandlers)
{
return;
}
var buffer = _receivedPacketBuffer.ToArray();
_receivedPacketBuffer.SetLength(0);
InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
}
public void BeginSendPacket(MqttPacketBuffer buffer)
{
if (!_asyncEvent.HasHandlers)
{
return;
return CompletedTask.Instance;
}
// Create a copy of the actual packet so that the inspector gets no access
@@ -65,7 +57,20 @@ namespace MQTTnet.Adapter
// intended for debugging etc. so that this is OK.
var bufferCopy = buffer.ToArray();
InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound);
return InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound);
}
public Task EndReceivePacket()
{
if (!_asyncEvent.HasHandlers)
{
return CompletedTask.Instance;
}
var buffer = _receivedPacketBuffer.ToArray();
_receivedPacketBuffer.SetLength(0);
return InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
}
public void FillReceiveBuffer(byte[] buffer)
@@ -74,16 +79,16 @@ namespace MQTTnet.Adapter
{
return;
}
_receivedPacketBuffer?.Write(buffer, 0, buffer.Length);
}
void InspectPacket(byte[] buffer, MqttPacketFlowDirection direction)
async Task InspectPacket(byte[] buffer, MqttPacketFlowDirection direction)
{
try
{
var eventArgs = new InspectMqttPacketEventArgs(direction, buffer);
_asyncEvent.InvokeAsync(eventArgs).GetAwaiter().GetResult();
await _asyncEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
catch (Exception exception)
{
+24 -18
View File
@@ -309,6 +309,7 @@ namespace MQTTnet.Client
AuthenticationMethod = Options.AuthenticationMethod,
AuthenticationData = data.AuthenticationData,
ReasonString = data.ReasonString,
ReasonCode = data.ReasonCode,
UserProperties = data.UserProperties
};
@@ -708,12 +709,15 @@ namespace MQTTnet.Client
await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false);
}
}
catch (ObjectDisposedException)
{
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error(exception, "Error while handling application message.");
_logger.Error(exception, "Error while handling application message");
}
}
}
@@ -787,7 +791,7 @@ namespace MQTTnet.Client
{
try
{
_logger.Verbose("Start receiving packets.");
_logger.Verbose("Start receiving packets");
while (!cancellationToken.IsCancellationRequested)
{
@@ -825,20 +829,21 @@ namespace MQTTnet.Client
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "Communication error while receiving packets.");
_logger.Warning(exception, "Communication error while receiving packets");
}
else
{
_logger.Error(exception, "Error while receiving packets.");
_logger.Error(exception, "Error while receiving packets");
}
_packetDispatcher.FailAll(exception);
// The packet dispatcher is set to null when the client is being disposed so it may already being gone!
_packetDispatcher?.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
finally
{
_logger.Verbose("Stopped receiving packets.");
_logger.Verbose("Stopped receiving packets");
}
}
@@ -860,7 +865,7 @@ namespace MQTTnet.Client
}
catch (Exception exception)
{
_logger.Warning(exception, "Error when sending request packet ({0}).", requestPacket.GetType().Name);
_logger.Warning(exception, "Error when sending {0} request packet", requestPacket.GetType().Name);
packetAwaitable.Fail(exception);
}
@@ -872,7 +877,7 @@ namespace MQTTnet.Client
{
if (exception is MqttCommunicationTimedOutException)
{
_logger.Warning("Timeout while waiting for response packet ({0}).", typeof(TResponsePacket).Name);
_logger.Warning("Timeout while waiting for {0} response packet", typeof(TResponsePacket).Name);
}
throw;
@@ -925,7 +930,7 @@ namespace MQTTnet.Client
static void ThrowNotConnected()
{
throw new MqttCommunicationException("The client is not connected.");
throw new MqttClientNotConnectedException();
}
void TryInitiateDisconnect()
@@ -938,7 +943,7 @@ namespace MQTTnet.Client
}
catch (Exception exception)
{
_logger.Warning(exception, "Error while initiating disconnect.");
_logger.Warning(exception, "Error while initiating disconnect");
}
}
}
@@ -973,7 +978,7 @@ namespace MQTTnet.Client
}
else if (packet is MqttPingReqPacket)
{
throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a Client to the Server only.");
throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a client to the server only.");
}
else
{
@@ -995,14 +1000,15 @@ namespace MQTTnet.Client
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "Communication error while receiving packets.");
_logger.Warning(exception, "Communication error while receiving packets");
}
else
{
_logger.Error(exception, $"Error while processing received packet ({packet.GetType().Name}).");
_logger.Error(exception, "Error while processing received {0} packet", packet.GetType().Name);
}
_packetDispatcher.FailAll(exception);
// The packet dispatcher may already be gone due to disconnect etc!
_packetDispatcher?.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
@@ -1012,7 +1018,7 @@ namespace MQTTnet.Client
{
try
{
_logger.Verbose("Start sending keep alive packets.");
_logger.Verbose("Start sending keep alive packets");
var keepAlivePeriod = Options.KeepAlivePeriod;
@@ -1050,18 +1056,18 @@ namespace MQTTnet.Client
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "Communication error while sending/receiving keep alive packets.");
_logger.Warning(exception, "Communication error while sending/receiving keep alive packets");
}
else
{
_logger.Error(exception, "Error exception while sending/receiving keep alive packets.");
_logger.Error(exception, "Error exception while sending/receiving keep alive packets");
}
await DisconnectInternal(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
}
finally
{
_logger.Verbose("Stopped sending keep alive packets.");
_logger.Verbose("Stopped sending keep alive packets");
}
}
}
@@ -2,9 +2,12 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
#pragma warning disable CS0612 // Type or member is obsolete
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using MQTTnet.Formatter;
using MQTTnet.Packets;
@@ -20,7 +23,9 @@ namespace MQTTnet.Client
MqttClientTcpOptions _tcpOptions;
MqttClientTlsOptions _tlsOptions;
EndPoint _remoteEndPoint;
int? _port;
[Obsolete] MqttClientOptionsBuilderTlsParameters _tlsParameters;
MqttClientWebSocketOptions _webSocketOptions;
@@ -37,7 +42,7 @@ namespace MQTTnet.Client
// on the order of called methods from the builder).
// The builder prefers the explicitly set TLS options!
var tlsOptions = _tlsOptions ?? _tcpOptions?.TlsOptions;
if (_tlsParameters != null)
{
if (_tlsParameters?.UseTls == true)
@@ -61,6 +66,31 @@ namespace MQTTnet.Client
if (_tcpOptions != null)
{
_tcpOptions.TlsOptions = tlsOptions;
if (_remoteEndPoint == null)
{
throw new ArgumentException("No endpoint is set.");
}
if (_remoteEndPoint is DnsEndPoint dns)
{
if (dns.Port == 0)
{
if (_port.HasValue)
{
_remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value);
}
else
{
_remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure);
}
}
}
if (_tcpOptions.RemoteEndpoint == null)
{
_tcpOptions.RemoteEndpoint = _remoteEndPoint;
}
}
else if (_webSocketOptions != null)
{
@@ -305,11 +335,20 @@ namespace MQTTnet.Client
public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null)
{
_tcpOptions = new MqttClientTcpOptions
{
Server = server,
Port = port
};
_tcpOptions = new MqttClientTcpOptions();
// The value 0 will be updated when building the options.
// This a backward compatibility feature.
_remoteEndPoint = new DnsEndPoint(server, port ?? 0);
_port = port;
return this;
}
public MqttClientOptionsBuilder WithEndPoint(EndPoint endPoint)
{
_remoteEndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint));
_tcpOptions = new MqttClientTcpOptions();
return this;
}
@@ -11,34 +11,32 @@ namespace MQTTnet.Client
{
public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
/// <summary>
/// Gets the local endpoint (network card) which is used by the client.
/// Set it to _null_ to let the OS select the network card.
/// </summary>
public EndPoint LocalEndpoint { get; set; }
public int BufferSize { get; set; } = 8192;
/// <summary>
/// Gets or sets whether the underlying socket should run in dual mode.
/// Leaving this _null_ will avoid setting this value at socket level.
/// Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
/// Gets or sets whether the underlying socket should run in dual mode.
/// Leaving this _null_ will avoid setting this value at socket level.
/// Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
/// </summary>
public bool? DualMode { get; set; }
public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
/// <summary>
/// Gets the local endpoint (network card) which is used by the client.
/// Set it to _null_ to let the OS select the network card.
/// </summary>
public EndPoint LocalEndpoint { get; set; }
public bool NoDelay { get; set; } = true;
public int? Port { get; set; }
public string Server { get; set; }
public EndPoint RemoteEndpoint { get; set; }
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
public override string ToString()
{
return Server + ":" + this.GetPort();
return RemoteEndpoint?.ToString() ?? string.Empty;
}
}
}
@@ -1,23 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
namespace MQTTnet.Client
{
public static class MqttClientTcpOptionsExtensions
{
public static int GetPort(this MqttClientTcpOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (options.Port.HasValue)
{
return options.Port.Value;
}
return !(options.TlsOptions?.UseTls ?? false) ? 1883 : 8883;
}
}
}
@@ -50,8 +50,13 @@ namespace MQTTnet.Client
#if NET48 || NETCOREAPP3_1_OR_GREATER
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls13;
#else
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | (SslProtocols)0x00003000 /*Tls13*/;
#endif
#if NET7_0_OR_GREATER
public X509Certificate2Collection TrustChain { get; set; }
#endif
}
}
@@ -131,6 +131,15 @@ namespace MQTTnet.Client
_tlsOptions.EncryptionPolicy = encryptionPolicy;
return this;
}
#endif
#if NET7_0_OR_GREATER
public MqttClientTlsOptionsBuilder WithTrustChain(X509Certificate2Collection chain)
{
_tlsOptions.TrustChain = chain;
return this;
}
#endif
}
}
@@ -0,0 +1,19 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
namespace MQTTnet.Exceptions
{
public class MqttClientNotConnectedException : MqttCommunicationException
{
public MqttClientNotConnectedException() : base("The MQTT client is not connected.")
{
}
public MqttClientNotConnectedException(Exception innerException) : base("The MQTT client is not connected.", innerException)
{
}
}
}
@@ -5,45 +5,107 @@
using MQTTnet.Client;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Formatter
{
public sealed class MqttDisconnectPacketFactory
{
readonly MqttDisconnectPacket _normalDisconnection = new MqttDisconnectPacket
static readonly MqttDisconnectPacket DefaultNormalDisconnection = new MqttDisconnectPacket
{
ReasonCode = MqttDisconnectReasonCode.NormalDisconnection
ReasonCode = MqttDisconnectReasonCode.NormalDisconnection,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
};
static readonly MqttDisconnectPacket DefaultServerShuttingDown = new MqttDisconnectPacket
{
ReasonCode = MqttDisconnectReasonCode.ServerShuttingDown,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
};
static readonly MqttDisconnectPacket DefaultUnspecifiedError = new MqttDisconnectPacket
{
ReasonCode = MqttDisconnectReasonCode.UnspecifiedError,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
};
public MqttDisconnectPacket Create(MqttDisconnectReasonCode reasonCode)
{
if (reasonCode == MqttDisconnectReasonCode.NormalDisconnection)
{
return _normalDisconnection;
return DefaultNormalDisconnection;
}
if (reasonCode == MqttDisconnectReasonCode.ServerShuttingDown)
{
return DefaultServerShuttingDown;
}
if (reasonCode == MqttDisconnectReasonCode.UnspecifiedError)
{
return DefaultUnspecifiedError;
}
return new MqttDisconnectPacket
{
ReasonCode = reasonCode
ReasonCode = reasonCode,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
};
}
public MqttDisconnectPacket Create(MqttServerStopOptions serverStopOptions)
{
if (serverStopOptions == null)
{
return DefaultServerShuttingDown;
}
return Create(serverStopOptions.DefaultClientDisconnectOptions);
}
public MqttDisconnectPacket Create(MqttServerClientDisconnectOptions clientDisconnectOptions)
{
if (clientDisconnectOptions == null)
{
return DefaultNormalDisconnection;
}
return new MqttDisconnectPacket
{
ReasonCode = clientDisconnectOptions.ReasonCode,
UserProperties = clientDisconnectOptions.UserProperties,
ReasonString = clientDisconnectOptions.ReasonString,
ServerReference = clientDisconnectOptions.ServerReference,
SessionExpiryInterval = 0 // TODO: Not yet supported!
};
}
public MqttDisconnectPacket Create(MqttClientDisconnectOptions clientDisconnectOptions)
{
var packet = new MqttDisconnectPacket();
if (clientDisconnectOptions == null)
{
packet.ReasonCode = MqttDisconnectReasonCode.NormalDisconnection;
}
else
{
packet.ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason;
packet.UserProperties = clientDisconnectOptions.UserProperties;
packet.SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval;
return DefaultNormalDisconnection;
}
return packet;
return new MqttDisconnectPacket
{
ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason,
UserProperties = clientDisconnectOptions.UserProperties,
SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval
};
}
}
}
@@ -279,7 +279,7 @@ namespace MQTTnet.Formatter.V5
_propertiesWriter.WriteReasonString(packet.ReasonString);
_propertiesWriter.WriteUserProperties(packet.UserProperties);
if (_bufferWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success)
if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success)
{
_bufferWriter.WriteByte((byte)packet.ReasonCode);
_propertiesWriter.WriteTo(_bufferWriter);
@@ -375,7 +375,7 @@ namespace MQTTnet.Formatter.V5
_bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier);
if (_bufferWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success)
if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success)
{
_bufferWriter.WriteByte((byte)packet.ReasonCode);
_propertiesWriter.WriteTo(_bufferWriter);
@@ -534,4 +534,4 @@ namespace MQTTnet.Formatter.V5
}
}
}
}
}
@@ -170,11 +170,16 @@ namespace MQTTnet.Implementations
_socket.Bind(localEndPoint);
}
public async Task ConnectAsync(string host, int port, CancellationToken cancellationToken)
public Task ConnectAsync(string host, int port, CancellationToken cancellationToken)
{
if (host is null)
return ConnectAsync(new DnsEndPoint(host, port), cancellationToken);
}
public async Task ConnectAsync(EndPoint endPoint, CancellationToken cancellationToken)
{
if (endPoint is null)
{
throw new ArgumentNullException(nameof(host));
throw new ArgumentNullException(nameof(endPoint));
}
cancellationToken.ThrowIfCancellationRequested();
@@ -191,15 +196,15 @@ namespace MQTTnet.Implementations
#endif
#if NET5_0_OR_GREATER
await _socket.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false);
await _socket.ConnectAsync(endPoint, cancellationToken).ConfigureAwait(false);
#else
// Workaround for: https://github.com/dotnet/corefx/issues/24430
using (cancellationToken.Register(_socketDisposeAction))
{
#if NET452 || NET461
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, host, port, null).ConfigureAwait(false);
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false);
#else
await _socket.ConnectAsync(host, port).ConfigureAwait(false);
await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
#endif
}
#endif
@@ -217,7 +222,7 @@ namespace MQTTnet.Implementations
throw new MqttCommunicationTimedOutException();
}
throw new MqttCommunicationException($"Error while connecting with host '{host}:{port}'.", socketException);
throw new MqttCommunicationException($"Error while connecting host '{endPoint}'.", socketException);
}
catch (ObjectDisposedException)
{
@@ -17,6 +17,7 @@ using MQTTnet.Channel;
using MQTTnet.Client;
using MQTTnet.Server;
using System.Runtime.InteropServices.WindowsRuntime;
using System.Net;
namespace MQTTnet.Implementations
{
@@ -65,9 +66,22 @@ namespace MQTTnet.Implementations
_socket.Control.KeepAlive = true;
}
string hostName;
string serviceName;
if (_options.RemoteEndpoint is DnsEndPoint dns)
{
hostName = dns.Host;
serviceName = dns.Port.ToString();
}
else
{
throw new NotSupportedException("UWP only supports a DNS endpoint.");
}
if (_options.TlsOptions?.UseTls != true)
{
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString()).AsTask().ConfigureAwait(false);
await _socket.ConnectAsync(new HostName(hostName), serviceName).AsTask().ConfigureAwait(false);
}
else
{
@@ -88,7 +102,7 @@ namespace MQTTnet.Implementations
socketProtectionLevel = SocketProtectionLevel.Tls10;
}
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), socketProtectionLevel).AsTask().ConfigureAwait(false);
await _socket.ConnectAsync(new HostName(hostName), serviceName, socketProtectionLevel).AsTask().ConfigureAwait(false);
}
Endpoint = _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;
@@ -5,6 +5,7 @@
#if !WINDOWS_UWP
using System;
using System.IO;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Runtime.ExceptionServices;
@@ -92,7 +93,7 @@ namespace MQTTnet.Implementations
socket.DualMode = _tcpOptions.DualMode.Value;
}
await socket.ConnectAsync(_tcpOptions.Server, _tcpOptions.GetPort(), cancellationToken).ConfigureAwait(false);
await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
@@ -103,9 +104,11 @@ namespace MQTTnet.Implementations
var targetHost = _tcpOptions.TlsOptions.TargetHost;
if (string.IsNullOrEmpty(targetHost))
{
targetHost = _tcpOptions.Server;
if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
{
targetHost = dns.Host;
}
}
var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
try
{
@@ -115,14 +118,31 @@ namespace MQTTnet.Implementations
ApplicationProtocols = _tcpOptions.TlsOptions.ApplicationProtocols,
ClientCertificates = LoadCertificates(),
EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol,
CertificateRevocationCheckMode =
_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
CertificateRevocationCheckMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors
? X509RevocationMode.NoCheck
: _tcpOptions.TlsOptions.RevocationMode,
TargetHost = targetHost,
CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy,
EncryptionPolicy = _tcpOptions.TlsOptions.EncryptionPolicy,
AllowRenegotiation = _tcpOptions.TlsOptions.AllowRenegotiation
};
#if NET7_0_OR_GREATER
if (_tcpOptions.TlsOptions.TrustChain?.Count > 0)
{
sslOptions.CertificateChainPolicy = new X509ChainPolicy
{
TrustMode = X509ChainTrustMode.CustomRootTrust,
VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors
? X509RevocationMode.NoCheck
: _tcpOptions.TlsOptions.RevocationMode
};
sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
}
#endif
await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
#else
await sslStream.AuthenticateAsClientAsync(
+7 -2
View File
@@ -104,17 +104,22 @@ namespace MQTTnet.Internal
_signal.Dispose();
}
public void DropFirstItem(MqttPacketBusPartition partition)
public MqttPacketBusItem DropFirstItem(MqttPacketBusPartition partition)
{
lock (_syncRoot)
{
var partitionInstance = _partitions[(int)partition];
if (partitionInstance.Any())
if (partitionInstance.Count > 0)
{
var firstItem = partitionInstance.First.Value;
partitionInstance.RemoveFirst();
return firstItem;
}
}
return null;
}
public void EnqueueItem(MqttPacketBusItem item, MqttPacketBusPartition partition)
@@ -19,7 +19,6 @@ namespace MQTTnet.LowLevelClient
readonly IMqttClientAdapterFactory _clientAdapterFactory;
readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
readonly MqttNetSourceLogger _logger;
readonly IMqttNetLogger _rootLogger;
IMqttChannelAdapter _adapter;
+3 -4
View File
@@ -23,6 +23,7 @@
<Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<DelaySign>false</DelaySign>
<SignAssembly>false</SignAssembly>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -69,10 +70,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -17,4 +17,5 @@
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cmanagement/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Coptions/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstatus/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstopping/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstorage/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+11
View File
@@ -11,6 +11,7 @@ using MQTTnet.Diagnostics;
using MQTTnet.Implementations;
using MQTTnet.LowLevelClient;
using MQTTnet.Server;
using MQTTnet.Server.Disconnecting;
using MqttClient = MQTTnet.Client.MqttClient;
namespace MQTTnet
@@ -175,6 +176,16 @@ namespace MQTTnet
return new MqttServer(options, serverAdapters, DefaultLogger);
}
public MqttServerClientDisconnectOptionsBuilder CreateMqttServerClientDisconnectOptionsBuilder()
{
return new MqttServerClientDisconnectOptionsBuilder();
}
public MqttServerStopOptionsBuilder CreateMqttServerStopOptionsBuilder()
{
return new MqttServerStopOptionsBuilder();
}
public MqttServerOptionsBuilder CreateServerOptionsBuilder()
{
return new MqttServerOptionsBuilder();
+13
View File
@@ -0,0 +1,13 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
namespace MQTTnet.Protocol
{
public static class MqttPorts
{
public const int Default = 1883;
public const int Secure = 8883;
}
}
@@ -0,0 +1,33 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server.Disconnecting
{
public sealed class MqttServerClientDisconnectOptions
{
public MqttDisconnectReasonCode ReasonCode { get; set; } = MqttDisconnectReasonCode.NormalDisconnection;
/// <summary>
/// The reason string is sent to every client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ReasonString { get; set; }
/// <summary>
/// The server reference is sent to every client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ServerReference { get; set; }
/// <summary>
/// These user properties are sent to every client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties { get; set; }
}
}
@@ -0,0 +1,55 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server.Disconnecting
{
public sealed class MqttServerClientDisconnectOptionsBuilder
{
readonly MqttServerClientDisconnectOptions _options = new MqttServerClientDisconnectOptions();
public MqttServerClientDisconnectOptions Build()
{
return _options;
}
public MqttServerClientDisconnectOptionsBuilder WithReasonCode(MqttDisconnectReasonCode value)
{
_options.ReasonCode = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithReasonString(string value)
{
_options.ReasonString = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithServerReference(string value)
{
_options.ServerReference = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithUserProperties(List<MqttUserProperty> value)
{
_options.UserProperties = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, string value)
{
if (_options.UserProperties == null)
{
_options.UserProperties = new List<MqttUserProperty>();
}
_options.UserProperties.Add(new MqttUserProperty(name, value));
return this;
}
}
}
@@ -0,0 +1,27 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
namespace MQTTnet.Server
{
public sealed class ApplicationMessageEnqueuedEventArgs : EventArgs
{
public ApplicationMessageEnqueuedEventArgs(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage, bool isDropped)
{
SenderClientId = senderClientId ?? throw new ArgumentNullException( nameof(senderClientId));
ReceiverClientId = receiverClientId ?? throw new ArgumentNullException(nameof(receiverClientId));
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
IsDropped = isDropped;
}
public string SenderClientId { get; }
public string ReceiverClientId { get; }
public bool IsDropped { get; }
public MqttApplicationMessage ApplicationMessage { get; }
}
}
@@ -0,0 +1,22 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class QueueMessageOverwrittenEventArgs : EventArgs
{
public QueueMessageOverwrittenEventArgs(string receiverClientId, MqttPacket packet)
{
ReceiverClientId = receiverClientId ?? throw new ArgumentNullException(nameof(receiverClientId));
Packet = packet ?? throw new ArgumentNullException(nameof(packet));
}
public MqttPacket Packet { get; }
public string ReceiverClientId { get; }
}
}
@@ -0,0 +1,12 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
namespace MQTTnet.Server
{
public enum EnqueueDataPacketResult
{
Enqueued,
Dropped
}
}
+24 -17
View File
@@ -4,6 +4,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
@@ -14,6 +15,7 @@ using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
@@ -95,15 +97,16 @@ namespace MQTTnet.Server
{
var cancellationToken = _cancellationToken.Token;
IsRunning = true;
_ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
_ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default)
.ConfigureAwait(false);
await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
}
finally
{
IsRunning = false;
Session.DisconnectedTimestamp = DateTime.UtcNow;
_cancellationToken?.TryCancel();
@@ -141,7 +144,7 @@ namespace MQTTnet.Server
Statistics.HandleSentPacket(packet);
}
public async Task StopAsync(MqttDisconnectReasonCode reason)
public async Task StopAsync(MqttServerClientDisconnectOptions disconnectOptions)
{
IsRunning = false;
@@ -150,16 +153,20 @@ namespace MQTTnet.Server
// Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
{
// The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
// This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
// a configuration option is requested in the future.
if (reason != MqttDisconnectReasonCode.NormalDisconnection)
// From RFC: The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
// This library does not sent a DISCONNECT packet for a normal disconnection.
// TODO: Maybe adding a configuration option is requested in the future.
if (disconnectOptions != null)
{
// Is is very important to send the DISCONNECT packet here BEFORE cancelling the
// token because the entire connection is closed (disposed) as soon as the cancellation
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
// at the client!
await TrySendDisconnectPacket(reason).ConfigureAwait(false);
if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
!string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
{
// Is is very important to send the DISCONNECT packet here BEFORE cancelling the
// token because the entire connection is closed (disposed) as soon as the cancellation
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
// at the client!
await TrySendDisconnectPacket(disconnectOptions).ConfigureAwait(false);
}
}
}
}
@@ -219,7 +226,7 @@ namespace MQTTnet.Server
if (dispatchApplicationMessageResult.CloseConnection)
{
await StopAsync(MqttDisconnectReasonCode.UnspecifiedError);
await StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.UnspecifiedError });
return;
}
@@ -550,14 +557,14 @@ namespace MQTTnet.Server
_cancellationToken?.TryCancel();
}
async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode)
async Task TrySendDisconnectPacket(MqttServerClientDisconnectOptions options)
{
try
{
// This also indicates that it was tried at least!
_disconnectPacketSent = true;
var disconnectPacket = MqttPacketFactories.Disconnect.Create(reasonCode);
var disconnectPacket = MqttPacketFactories.Disconnect.Create(options);
using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
{
@@ -566,7 +573,7 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, reasonCode);
_logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
}
}
}
@@ -16,6 +16,7 @@ using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
@@ -32,11 +33,11 @@ namespace MQTTnet.Server
readonly MqttRetainedMessagesManager _retainedMessagesManager;
readonly IMqttNetLogger _rootLogger;
readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
// The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only.
// See the MqttSubscription object for a detailed explanation.
readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage();
readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>();
public MqttClientSessionsManager(
@@ -58,8 +59,13 @@ namespace MQTTnet.Server
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
}
public async Task CloseAllConnections()
public async Task CloseAllConnections(MqttServerClientDisconnectOptions options)
{
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
List<MqttClient> connections;
lock (_clients)
{
@@ -69,7 +75,7 @@ namespace MQTTnet.Server
foreach (var connection in connections)
{
await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
await connection.StopAsync(options).ConfigureAwait(false);
}
}
@@ -101,7 +107,7 @@ namespace MQTTnet.Server
{
if (connection != null)
{
await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
await connection.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.NormalDisconnection }).ConfigureAwait(false);
}
}
catch (Exception exception)
@@ -204,7 +210,7 @@ namespace MQTTnet.Server
{
continue;
}
if (_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
{
var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, session.Id, applicationMessage);
@@ -238,7 +244,14 @@ namespace MQTTnet.Server
matchingSubscribersCount++;
session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
var result = session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
if (_eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.HasHandlers)
{
var eventArgs = new ApplicationMessageEnqueuedEventArgs(senderId, session.Id, applicationMessage, result == EnqueueDataPacketResult.Dropped);
await _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
_logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'", session.Id, applicationMessage.Topic);
}
@@ -366,7 +379,8 @@ namespace MQTTnet.Server
if (_eventContainer.ClientConnectedEvent.HasHandlers)
{
var eventArgs = new ClientConnectedEventArgs(connectPacket,
var eventArgs = new ClientConnectedEventArgs(
connectPacket,
channelAdapter.PacketFormatterAdapter.ProtocolVersion,
channelAdapter.Endpoint,
client.Session.Items);
@@ -595,13 +609,13 @@ namespace MQTTnet.Server
session.IsPersistent = sessionShouldPersist;
session.DisconnectedTimestamp = null;
session.Recover();
connAckPacket.IsSessionPresent = true;
}
}
_sessionsStorage.UpdateSession(connectPacket.ClientId, session);
// Create a new client (always required).
lock (_clients)
{
@@ -631,7 +645,8 @@ namespace MQTTnet.Server
if (oldClient != null)
{
await oldClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
// TODO: Consider event here for session takeover to allow manipulation of user properties etc.
await oldClient.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.SessionTakenOver }).ConfigureAwait(false);
if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
{
@@ -734,4 +749,4 @@ namespace MQTTnet.Server
return eventArgs;
}
}
}
}
@@ -92,7 +92,7 @@ namespace MQTTnet.Server
}
}
private struct Statistics
struct Statistics
{
public long _receivedPacketsCount;
public long _sentPacketsCount;
@@ -22,6 +22,10 @@ namespace MQTTnet.Server
public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
public AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs> InterceptingClientEnqueueEvent { get; } = new AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs>();
public AsyncEvent<ApplicationMessageEnqueuedEventArgs> ApplicationMessageEnqueuedOrDroppedEvent { get; } = new AsyncEvent<ApplicationMessageEnqueuedEventArgs>();
public AsyncEvent<QueueMessageOverwrittenEventArgs> QueuedApplicationMessageOverwrittenEvent { get; } = new AsyncEvent<QueueMessageOverwrittenEventArgs>();
public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
@@ -8,11 +8,20 @@ using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public sealed class MqttServerKeepAliveMonitor
{
static readonly MqttServerClientDisconnectOptions DefaultDisconnectOptions = new MqttServerClientDisconnectOptions
{
ReasonCode = MqttDisconnectReasonCode.KeepAliveTimeout,
UserProperties = null,
ReasonString = null,
ServerReference = null
};
readonly MqttNetSourceLogger _logger;
readonly MqttServerOptions _options;
readonly MqttClientSessionsManager _sessionsManager;
@@ -43,12 +52,12 @@ namespace MQTTnet.Server
{
try
{
_logger.Info("Starting keep alive monitor.");
_logger.Info("Starting keep alive monitor");
while (!cancellationToken.IsCancellationRequested)
{
TryProcessClients();
Sleep(_options.KeepAliveMonitorInterval);
Sleep(_options.KeepAliveOptions.MonitorInterval);
}
}
catch (OperationCanceledException)
@@ -56,11 +65,11 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while checking keep alive timeouts.");
_logger.Error(exception, "Unhandled exception while checking keep alive timeouts");
}
finally
{
_logger.Verbose("Stopped checking keep alive timeout.");
_logger.Verbose("Stopped checking keep alive timeout");
}
}
@@ -98,13 +107,6 @@ namespace MQTTnet.Server
return;
}
if (connection.ChannelAdapter.IsReadingPacket)
{
// The connection is currently reading a (large) packet. So it is obviously
// doing something and thus "connected".
return;
}
// Values described here: [MQTT-3.1.2-24].
// If the client sends 5 sec. the server will allow up to 7.5 seconds.
// If the client sends 1 sec. the server will allow up to 1.5 seconds.
@@ -117,18 +119,18 @@ namespace MQTTnet.Server
return;
}
_logger.Warning("Client '{0}': Did not receive any packet or keep alive signal.", connection.Id);
_logger.Warning("Client '{0}': Did not receive any packet or keep alive signal", connection.Id);
// Execute the disconnection in background so that the keep alive monitor can continue
// with checking other connections.
// We do not need to wait for the task so no await is needed.
// Also the internal state of the connection must be swapped to "Finalizing" because the
// next iteration of the keep alive timer happens.
var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout);
_ = connection.StopAsync(DefaultDisconnectOptions);
}
catch (Exception exception)
{
_logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.Id);
_logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts", connection.Id);
}
}
+11 -3
View File
@@ -18,6 +18,7 @@ namespace MQTTnet.Server
public sealed class MqttSession : IDisposable
{
readonly MqttClientSessionsManager _clientSessionsManager;
readonly MqttServerEventContainer _eventContainer;
readonly MqttPacketBus _packetBus = new MqttPacketBus();
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
@@ -44,6 +45,7 @@ namespace MQTTnet.Server
_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
_clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
_subscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);
}
@@ -117,20 +119,25 @@ namespace MQTTnet.Server
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
}
public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
public EnqueueDataPacketResult EnqueueDataPacket(MqttPacketBusItem packetBusItem)
{
if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
{
if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
{
return;
return EnqueueDataPacketResult.Dropped;
}
if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
{
// Only drop from the data partition. Dropping from control partition might break the connection
// because the client does not receive PINGREQ packets etc. any longer.
_packetBus.DropFirstItem(MqttPacketBusPartition.Data);
var firstItem = _packetBus.DropFirstItem(MqttPacketBusPartition.Data);
if (firstItem != null && _eventContainer.QueuedApplicationMessageOverwrittenEvent.HasHandlers)
{
var eventArgs = new QueueMessageOverwrittenEventArgs(Id, firstItem.Packet);
_eventContainer.QueuedApplicationMessageOverwrittenEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
}
}
@@ -147,6 +154,7 @@ namespace MQTTnet.Server
}
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
return EnqueueDataPacketResult.Enqueued;
}
public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)
+56 -16
View File
@@ -14,6 +14,7 @@ using MQTTnet.Diagnostics;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
@@ -29,6 +30,7 @@ namespace MQTTnet.Server
readonly IMqttNetLogger _rootLogger;
CancellationTokenSource _cancellationTokenSource;
bool _isStopping;
public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
{
@@ -49,6 +51,12 @@ namespace MQTTnet.Server
_keepAliveMonitor = new MqttServerKeepAliveMonitor(options, _clientSessionsManager, _rootLogger);
}
public event Func<ApplicationMessageEnqueuedEventArgs, Task> ApplicationMessageEnqueuedOrDroppedAsync
{
add => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.AddHandler(value);
remove => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.RemoveHandler(value);
}
public event Func<ApplicationMessageNotConsumedEventArgs, Task> ApplicationMessageNotConsumedAsync
{
add => _eventContainer.ApplicationMessageNotConsumedEvent.AddHandler(value);
@@ -133,6 +141,12 @@ namespace MQTTnet.Server
remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value);
}
public event Func<QueueMessageOverwrittenEventArgs, Task> QueuedApplicationMessageOverwrittenAsync
{
add => _eventContainer.QueuedApplicationMessageOverwrittenEvent.AddHandler(value);
remove => _eventContainer.QueuedApplicationMessageOverwrittenEvent.RemoveHandler(value);
}
public event Func<RetainedMessageChangedEventArgs, Task> RetainedMessageChangedAsync
{
add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value);
@@ -169,6 +183,13 @@ namespace MQTTnet.Server
remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value);
}
/// <summary>
/// Gets or sets whether the server will accept new connections.
/// If not, the server will close the connection without any notification (DISCONNECT packet).
/// This feature can be used when the server is shutting down.
/// </summary>
public bool AcceptNewConnections { get; set; } = true;
public bool IsStarted => _cancellationTokenSource != null;
/// <summary>
@@ -184,16 +205,21 @@ namespace MQTTnet.Server
return _retainedMessagesManager?.ClearMessages() ?? CompletedTask.Instance;
}
public Task DisconnectClientAsync(string id, MqttDisconnectReasonCode reasonCode)
public Task DisconnectClientAsync(string id, MqttServerClientDisconnectOptions options)
{
if (id == null)
{
throw new ArgumentNullException(nameof(id));
}
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
ThrowIfNotStarted();
return _clientSessionsManager.GetClient(id).StopAsync(reasonCode);
return _clientSessionsManager.GetClient(id).StopAsync(options);
}
public Task<IList<MqttClientStatus>> GetClientsAsync()
@@ -203,13 +229,6 @@ namespace MQTTnet.Server
return _clientSessionsManager.GetClientsStatus();
}
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
{
ThrowIfNotStarted();
return _retainedMessagesManager.GetMessages();
}
public Task<MqttApplicationMessage> GetRetainedMessageAsync(string topic)
{
if (topic == null)
@@ -222,6 +241,13 @@ namespace MQTTnet.Server
return _retainedMessagesManager.GetMessage(topic);
}
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
{
ThrowIfNotStarted();
return _retainedMessagesManager.GetMessages();
}
public Task<IList<MqttSessionStatus>> GetSessionsAsync()
{
ThrowIfNotStarted();
@@ -247,7 +273,7 @@ namespace MQTTnet.Server
if (string.IsNullOrEmpty(injectedApplicationMessage.ApplicationMessage.Topic))
{
throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported.");
throw new NotSupportedException("Injected application messages must contain a topic (topic alias is not supported)");
}
var sessionItems = injectedApplicationMessage.CustomSessionItems ?? ServerSessionItems;
@@ -263,6 +289,8 @@ namespace MQTTnet.Server
{
ThrowIfStarted();
_isStopping = false;
_cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = _cancellationTokenSource.Token;
@@ -278,11 +306,16 @@ namespace MQTTnet.Server
await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
_logger.Info("Started.");
_logger.Info("Started");
}
public async Task StopAsync()
public async Task StopAsync(MqttServerStopOptions options)
{
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
try
{
if (_cancellationTokenSource == null)
@@ -290,9 +323,11 @@ namespace MQTTnet.Server
return;
}
_isStopping = true;
_cancellationTokenSource.Cancel(false);
await _clientSessionsManager.CloseAllConnections().ConfigureAwait(false);
await _clientSessionsManager.CloseAllConnections(options.DefaultClientDisconnectOptions).ConfigureAwait(false);
foreach (var adapter in _adapters)
{
@@ -308,7 +343,7 @@ namespace MQTTnet.Server
await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
_logger.Info("Stopped.");
_logger.Info("Stopped");
}
public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
@@ -362,14 +397,14 @@ namespace MQTTnet.Server
ThrowIfDisposed();
ThrowIfNotStarted();
return _retainedMessagesManager?.UpdateMessage(null, retainedMessage);
return _retainedMessagesManager?.UpdateMessage(string.Empty, retainedMessage);
}
protected override void Dispose(bool disposing)
{
if (disposing)
{
StopAsync().GetAwaiter().GetResult();
StopAsync(new MqttServerStopOptions()).GetAwaiter().GetResult();
foreach (var adapter in _adapters)
{
@@ -382,6 +417,11 @@ namespace MQTTnet.Server
Task OnHandleClient(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{
if (_isStopping || !AcceptNewConnections)
{
return CompletedTask.Instance;
}
return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter, cancellationToken);
}
@@ -8,11 +8,22 @@ using System.Threading.Tasks;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public static class MqttServerExtensions
{
public static Task DisconnectClientAsync(this MqttServer server, string id, MqttDisconnectReasonCode reasonCode = MqttDisconnectReasonCode.NormalDisconnection)
{
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
return server.DisconnectClientAsync(id, new MqttServerClientDisconnectOptions { ReasonCode = reasonCode });
}
public static Task InjectApplicationMessage(
this MqttServer server,
string topic,
@@ -47,6 +58,16 @@ namespace MQTTnet.Server
}));
}
public static Task StopAsync(this MqttServer server)
{
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
return server.StopAsync(new MqttServerStopOptions());
}
public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
{
if (server == null)
@@ -0,0 +1,21 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
namespace MQTTnet.Server
{
public sealed class MqttServerKeepAliveOptions
{
/// <summary>
/// When this mode is enabled the MQTT server will not close a connection when the
/// client is currently sending a (large) payload. This may lead to "dead" connections
/// When this mode is disabled the MQTT server will disconnect a client when the keep
/// alive timeout is reached even if the client is currently sending a (large) payload.
/// </summary>
public bool DisconnectClientWhenReadingPayload { get; set; }
public TimeSpan MonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
}
}
@@ -9,12 +9,19 @@ namespace MQTTnet.Server
public sealed class MqttServerOptions
{
public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(100);
public MqttServerTcpEndpointOptions DefaultEndpointOptions { get; } = new MqttServerTcpEndpointOptions();
public bool EnablePersistentSessions { get; set; }
public TimeSpan KeepAliveMonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
[Obsolete("Use KeepAliveOptions instead.")]
public TimeSpan KeepAliveMonitorInterval
{
get => KeepAliveOptions.MonitorInterval;
set => KeepAliveOptions.MonitorInterval = value;
}
public MqttServerKeepAliveOptions KeepAliveOptions { get; } = new MqttServerKeepAliveOptions();
public int MaxPendingMessagesPerClient { get; set; } = 250;
@@ -5,7 +5,7 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Formatter;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
@@ -18,41 +18,46 @@ namespace MQTTnet.Server
_client = client ?? throw new ArgumentNullException(nameof(client));
}
/// <summary>
/// Gets or sets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary>
public string Id => _client.Id;
public long BytesReceived => _client.ChannelAdapter.BytesReceived;
public long BytesSent => _client.ChannelAdapter.BytesSent;
public DateTime ConnectedTimestamp => _client.Statistics.ConnectedTimestamp;
public string Endpoint => _client.Endpoint;
public MqttProtocolVersion ProtocolVersion => _client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
/// <summary>
/// Gets or sets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary>
public string Id => _client.Id;
public DateTime ConnectedTimestamp => _client.Statistics.ConnectedTimestamp;
public DateTime LastNonKeepAlivePacketReceivedTimestamp => _client.Statistics.LastNonKeepAlivePacketReceivedTimestamp;
public DateTime LastPacketReceivedTimestamp => _client.Statistics.LastPacketReceivedTimestamp;
public DateTime LastPacketSentTimestamp => _client.Statistics.LastPacketSentTimestamp;
public DateTime LastNonKeepAlivePacketReceivedTimestamp => _client.Statistics.LastNonKeepAlivePacketReceivedTimestamp;
public MqttProtocolVersion ProtocolVersion => _client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
public long ReceivedApplicationMessagesCount => _client.Statistics.ReceivedApplicationMessagesCount;
public long SentApplicationMessagesCount => _client.Statistics.SentApplicationMessagesCount;
public long ReceivedPacketsCount => _client.Statistics.ReceivedPacketsCount;
public long SentApplicationMessagesCount => _client.Statistics.SentApplicationMessagesCount;
public long SentPacketsCount => _client.Statistics.SentPacketsCount;
public MqttSessionStatus Session { get; set; }
public long BytesSent => _client.ChannelAdapter.BytesSent;
public long BytesReceived => _client.ChannelAdapter.BytesReceived;
public Task DisconnectAsync()
public Task DisconnectAsync(MqttServerClientDisconnectOptions options)
{
return _client.StopAsync(MqttDisconnectReasonCode.NormalDisconnection);
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
return _client.StopAsync(options);
}
public void ResetStatistics()
@@ -60,4 +65,4 @@ namespace MQTTnet.Server
_client.ResetStatistics();
}
}
}
}
@@ -0,0 +1,32 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Threading.Tasks;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public static class MqttClientStatusExtensions
{
static readonly MqttServerClientDisconnectOptions DefaultDisconnectOptions = new MqttServerClientDisconnectOptions
{
ReasonCode = MqttDisconnectReasonCode.NormalDisconnection,
ReasonString = null,
UserProperties = null,
ServerReference = null
};
public static Task DisconnectAsync(this MqttClientStatus clientStatus)
{
if (clientStatus == null)
{
throw new ArgumentNullException(nameof(clientStatus));
}
return clientStatus.DisconnectAsync(DefaultDisconnectOptions);
}
}
}
@@ -0,0 +1,24 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public sealed class MqttServerStopOptions
{
/// <summary>
/// These disconnect options are sent to every connected client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public MqttServerClientDisconnectOptions DefaultClientDisconnectOptions { get; set; } = new MqttServerClientDisconnectOptions
{
ReasonCode = MqttDisconnectReasonCode.ServerShuttingDown,
UserProperties = null,
ReasonString = null,
ServerReference = null
};
}
}
@@ -0,0 +1,39 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public sealed class MqttServerStopOptionsBuilder
{
readonly MqttServerStopOptions _options = new MqttServerStopOptions();
public MqttServerStopOptionsBuilder WithDefaultClientDisconnectOptions(MqttServerClientDisconnectOptions value)
{
_options.DefaultClientDisconnectOptions = value;
return this;
}
public MqttServerStopOptionsBuilder WithDefaultClientDisconnectOptions(Action<MqttServerClientDisconnectOptionsBuilder> builder)
{
if (builder == null)
{
throw new ArgumentNullException(nameof(builder));
}
var optionsBuilder = new MqttServerClientDisconnectOptionsBuilder();
builder.Invoke(optionsBuilder);
_options.DefaultClientDisconnectOptions = optionsBuilder.Build();
return this;
}
public MqttServerStopOptions Build()
{
return _options;
}
}
}