From fdf00f1b18dc61f9c2bd7afe8d24af5efe7f73ce Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Fri, 23 Dec 2022 17:09:02 +0100 Subject: [PATCH] More options for disconnect (#1557) * Expose more options in disconnect options. * Expose more options in disconnect API. * Refactor code. * Expose user properties in disconnected event. * Expose more disconnect options in server. * Fix build errors * Add new samples. * Fix client cleanup * Fix merge conflicts --- .github/workflows/ReleaseNotes.md | 3 + Samples/Client/Client_Connection_Samples.cs | 24 +- Samples/Server/Server_Diagnostics_Samples.cs | 53 +++++ .../LowLevelMqttClient_Tests.cs | 2 +- .../ManagedMqttClient_Tests.cs | 16 +- .../MqttClient/MqttClient_Connection_Tests.cs | 162 ++++++++++++++ .../MqttClient}/MqttClient_Tests.cs | 66 ++---- .../MqttClientOptionsBuilder_Tests.cs | 3 +- .../MQTTnet.Tests/Mockups/TestEnvironment.cs | 10 + Source/MQTTnet.Tests/Server/General.cs | 2 +- .../MqttClientConnectedEventArgs.cs | 2 +- .../MqttClientDisconnectOptions.cs | 17 +- .../MqttClientDisconnectOptionsBuilder.cs | 60 +++-- .../MqttClientDisconnectOptionsReason.cs | 27 +++ .../MqttClientDisconnectOptionsValidator.cs | 2 +- .../MqttClientDisconnectedEventArgs.cs | 42 +++- Source/MQTTnet/Client/MqttClient.cs | 205 +++++++++++------- Source/MQTTnet/Client/MqttClientExtensions.cs | 47 +++- .../Options/MqttClientOptionsBuilder.cs | 10 - .../MqttClientSubscribeOptionsBuilder.cs | 72 +++--- .../MqttClientUnsubscribeOptionsBuilder.cs | 79 +++---- .../Formatter/MqttDisconnectPacketFactory.cs | 2 + .../Implementations/CrossPlatformSocket.cs | 1 - Source/MQTTnet/Internal/TaskExtensions.cs | 44 +++- .../MQTTnet/MqttApplicationMessageBuilder.cs | 96 +------- .../Events/ClientDisconnectedEventArgs.cs | 43 +++- Source/MQTTnet/Server/Internal/MqttClient.cs | 15 +- .../Internal/MqttClientSessionsManager.cs | 13 +- 28 files changed, 752 insertions(+), 366 deletions(-) create mode 100644 Samples/Server/Server_Diagnostics_Samples.cs rename Source/MQTTnet.Tests/{Client => Clients/LowLevelMqttClient}/LowLevelMqttClient_Tests.cs (99%) rename Source/MQTTnet.Tests/{Client => Clients/ManagedMqttClient}/ManagedMqttClient_Tests.cs (95%) create mode 100644 Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs rename Source/MQTTnet.Tests/{Client => Clients/MqttClient}/MqttClient_Tests.cs (93%) rename Source/MQTTnet.Tests/{Client => Clients}/MqttClientOptionsBuilder_Tests.cs (93%) create mode 100644 Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsReason.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index abfab925..1a651a56 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1 +1,4 @@ +* [Client] Added support for passing MQTT v5 options (User properties etc.) for disconnects. +* [Client] An internal exception is no longer caught silently when calling _DisconnectAsync_ to indicate that the disconnect is not clean (BREAKING CHANGE). * [Client] MQTTv5 features are now checked and an exception is thrown if they are used when using protocol version 3.1.1 and lower. These checks can be disabled in client options. (BREAKING CHANGE!). +* [Server] Exposed MQTT v5 sent properties from the affected client in _ClientDisconnectedAsync_ event. diff --git a/Samples/Client/Client_Connection_Samples.cs b/Samples/Client/Client_Connection_Samples.cs index f42667b7..81808a4e 100644 --- a/Samples/Client/Client_Connection_Samples.cs +++ b/Samples/Client/Client_Connection_Samples.cs @@ -16,6 +16,28 @@ namespace MQTTnet.Samples.Client; public static class Client_Connection_Samples { + public static async Task Clean_Disconnect() + { + /* + * This sample disconnects in a clean way. This will send a MQTT DISCONNECT packet + * to the server and close the connection afterwards. + * + * See sample _Connect_Client_ for more details. + */ + + var mqttFactory = new MqttFactory(); + + using (var mqttClient = mqttFactory.CreateMqttClient()) + { + var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build(); + await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); + + // This will send the DISCONNECT packet. Calling _Dispose_ without DisconnectAsync the + // connection is closed in a "not clean" way. See MQTT specification for more details. + await mqttClient.DisconnectAsync(new MqttClientDisconnectOptionsBuilder().WithReason(MqttClientDisconnectOptionsReason.NormalDisconnection).Build()); + } + } + public static async Task Connect_Client() { /* @@ -232,7 +254,7 @@ public static class Client_Connection_Samples // Calling _DisconnectAsync_ will send a DISCONNECT packet before closing the connection. // Using a reason code requires MQTT version 5.0.0! - await mqttClient.DisconnectAsync(MqttClientDisconnectReason.ImplementationSpecificError); + await mqttClient.DisconnectAsync(MqttClientDisconnectOptionsReason.ImplementationSpecificError); } } diff --git a/Samples/Server/Server_Diagnostics_Samples.cs b/Samples/Server/Server_Diagnostics_Samples.cs new file mode 100644 index 00000000..0f31e9be --- /dev/null +++ b/Samples/Server/Server_Diagnostics_Samples.cs @@ -0,0 +1,53 @@ +// 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.Internal; +using MQTTnet.Packets; +using MQTTnet.Server; + +namespace MQTTnet.Samples.Server; + +// ReSharper disable UnusedType.Global +// ReSharper disable UnusedMember.Global +// ReSharper disable InconsistentNaming +// ReSharper disable EmptyConstructor +// ReSharper disable MemberCanBeMadeStatic.Local + +public static class Server_Diagnostics_Samples +{ + public static async Task Get_Notified_When_Client_Received_Message() + { + /* + * This sample starts a MQTT server and attaches an event which gets fired whenever + * a client received and acknowledged a PUBLISH packet. + */ + + var mqttFactory = new MqttFactory(); + var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build(); + + using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions)) + { + // Attach the event handler. + mqttServer.ClientAcknowledgedPublishPacketAsync += e => + { + Console.WriteLine($"Client '{e.ClientId}' acknowledged packet {e.PublishPacket.PacketIdentifier} with topic '{e.PublishPacket.Topic}'"); + + // It is also possible to read additional data from the client response. This requires casting the response packet. + var qos1AcknowledgePacket = e.AcknowledgePacket as MqttPubAckPacket; + Console.WriteLine($"QoS 1 reason code: {qos1AcknowledgePacket?.ReasonCode}"); + + var qos2AcknowledgePacket = e.AcknowledgePacket as MqttPubCompPacket; + Console.WriteLine($"QoS 2 reason code: {qos1AcknowledgePacket?.ReasonCode}"); + return CompletedTask.Instance; + }; + + await mqttServer.StartAsync(); + + Console.WriteLine("Press Enter to exit."); + Console.ReadLine(); + + await mqttServer.StopAsync(); + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/LowLevelMqttClient/LowLevelMqttClient_Tests.cs similarity index 99% rename from Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs rename to Source/MQTTnet.Tests/Clients/LowLevelMqttClient/LowLevelMqttClient_Tests.cs index e7366d3e..6e9485f2 100644 --- a/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/LowLevelMqttClient/LowLevelMqttClient_Tests.cs @@ -14,7 +14,7 @@ using MQTTnet.LowLevelClient; using MQTTnet.Packets; using MQTTnet.Protocol; -namespace MQTTnet.Tests.Client +namespace MQTTnet.Tests.Clients.LowLevelMqttClient { [TestClass] public sealed class LowLevelMqttClient_Tests : BaseTestClass diff --git a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs similarity index 95% rename from Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs rename to Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs index 69c9847b..f16d55fa 100644 --- a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs @@ -17,7 +17,7 @@ using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; -namespace MQTTnet.Tests.Client +namespace MQTTnet.Tests.Clients.ManagedMqttClient { [TestClass] public sealed class ManagedMqttClient_Tests : BaseTestClass @@ -164,7 +164,7 @@ namespace MQTTnet.Tests.Client .Build(); var dyingClient = testEnvironment.CreateClient(); - var dyingManagedClient = new ManagedMqttClient(dyingClient, testEnvironment.ClientLogger); + var dyingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(dyingClient, testEnvironment.ClientLogger); await dyingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build()); @@ -199,7 +199,7 @@ namespace MQTTnet.Tests.Client { var server = await testEnvironment.StartServer(); - var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetEventLogger()); + var managedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetEventLogger()); var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort); var connected = GetConnectedTask(managedClient); @@ -226,7 +226,7 @@ namespace MQTTnet.Tests.Client await testEnvironment.StartServer(); - var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetEventLogger()); + var managedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetEventLogger()); var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort); var storage = new ManagedMqttClientTestStorage(); @@ -439,7 +439,7 @@ namespace MQTTnet.Tests.Client } } - async Task CreateManagedClientAsync( + async Task CreateManagedClientAsync( TestEnvironment testEnvironment, IMqttClient underlyingClient = null, TimeSpan? connectionCheckInterval = null, @@ -456,7 +456,7 @@ namespace MQTTnet.Tests.Client // at connection check intervals managedOptions.ConnectionCheckInterval = connectionCheckInterval ?? TimeSpan.FromSeconds(0.1); - var managedClient = new ManagedMqttClient(underlyingClient ?? testEnvironment.CreateClient(), new MqttNetEventLogger()); + var managedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(underlyingClient ?? testEnvironment.CreateClient(), new MqttNetEventLogger()); var connected = GetConnectedTask(managedClient); @@ -473,7 +473,7 @@ namespace MQTTnet.Tests.Client /// name="managedClient" /> /// has connected /// - Task GetConnectedTask(ManagedMqttClient managedClient) + Task GetConnectedTask(MQTTnet.Extensions.ManagedClient.ManagedMqttClient managedClient) { var connected = new TaskCompletionSource(); @@ -495,7 +495,7 @@ namespace MQTTnet.Tests.Client /// name="expectedNumberOfMessages" /> /// have been received /// - Task> SetupReceivingOfMessages(ManagedMqttClient managedClient, int expectedNumberOfMessages) + Task> SetupReceivingOfMessages(MQTTnet.Extensions.ManagedClient.ManagedMqttClient managedClient, int expectedNumberOfMessages) { var receivedMessages = new List(); var result = new AsyncTaskCompletionSource>(); diff --git a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs new file mode 100644 index 00000000..e2ce36cf --- /dev/null +++ b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs @@ -0,0 +1,162 @@ +// 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; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Exceptions; +using MQTTnet.Formatter; +using MQTTnet.Internal; +using MQTTnet.Protocol; +using MQTTnet.Server; + +namespace MQTTnet.Tests.Clients.MqttClient +{ + [TestClass] + public sealed class MqttClient_Connection_Tests : BaseTestClass + { + [TestMethod] + [ExpectedException(typeof(MqttCommunicationException))] + public async Task Connect_To_Invalid_Server_Port_Not_Opened() + { + var client = new MqttFactory().CreateMqttClient(); + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5))) + { + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", 12345).Build(), timeout.Token); + } + } + + [TestMethod] + [ExpectedException(typeof(OperationCanceledException))] + public async Task Connect_To_Invalid_Server_Wrong_IP() + { + var client = new MqttFactory().CreateMqttClient(); + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2))) + { + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build(), timeout.Token); + } + } + + [TestMethod] + [ExpectedException(typeof(MqttCommunicationException))] + public async Task Connect_To_Invalid_Server_Wrong_Protocol() + { + var client = new MqttFactory().CreateMqttClient(); + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("http://127.0.0.1", 12345).WithTimeout(TimeSpan.FromSeconds(2)).Build()); + } + + [TestMethod] + public async Task ConnectTimeout_Throws_Exception() + { + var factory = new MqttFactory(); + using (var client = factory.CreateMqttClient()) + { + var disconnectHandlerCalled = false; + try + { + client.DisconnectedAsync += args => + { + disconnectHandlerCalled = true; + return CompletedTask.Instance; + }; + + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build()); + + Assert.Fail("Must fail!"); + } + catch (Exception exception) + { + Assert.IsNotNull(exception); + Assert.IsInstanceOfType(exception, typeof(MqttCommunicationException)); + } + + await LongTestDelay(); // disconnected handler is called async + Assert.IsTrue(disconnectHandlerCalled); + } + } + + [TestMethod] + public async Task Disconnect_Clean() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + ClientDisconnectedEventArgs eventArgs = null; + server.ClientDisconnectedAsync += args => + { + eventArgs = args; + return CompletedTask.Instance; + }; + + var client = await testEnvironment.ConnectClient(); + + var disconnectOptions = testEnvironment.Factory.CreateClientDisconnectOptionsBuilder().WithReason(MqttClientDisconnectOptionsReason.MessageRateTooHigh).Build(); + + // Perform a clean disconnect. + await client.DisconnectAsync(disconnectOptions); + + Assert.IsNotNull(eventArgs); + Assert.AreEqual(MqttClientDisconnectType.Clean, eventArgs.DisconnectType); + } + } + + [TestMethod] + public async Task Disconnect_Clean_With_Custom_Reason() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + var server = await testEnvironment.StartServer(); + + ClientDisconnectedEventArgs eventArgs = null; + server.ClientDisconnectedAsync += args => + { + eventArgs = args; + return CompletedTask.Instance; + }; + + var client = await testEnvironment.ConnectClient(); + + var disconnectOptions = testEnvironment.Factory.CreateClientDisconnectOptionsBuilder().WithReason(MqttClientDisconnectOptionsReason.MessageRateTooHigh).Build(); + + // Perform a clean disconnect. + await client.DisconnectAsync(disconnectOptions); + + Assert.IsNotNull(eventArgs); + Assert.AreEqual(MqttDisconnectReasonCode.MessageRateTooHigh, eventArgs.ReasonCode); + } + } + + [TestMethod] + public async Task Disconnect_Clean_With_User_Properties() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + var server = await testEnvironment.StartServer(); + + ClientDisconnectedEventArgs eventArgs = null; + server.ClientDisconnectedAsync += args => + { + eventArgs = args; + return CompletedTask.Instance; + }; + + var client = await testEnvironment.ConnectClient(); + + var disconnectOptions = testEnvironment.Factory.CreateClientDisconnectOptionsBuilder().WithUserProperty("test_name", "test_value").Build(); + + // Perform a clean disconnect. + await client.DisconnectAsync(disconnectOptions); + + Assert.IsNotNull(eventArgs); + Assert.IsNotNull(eventArgs.UserProperties); + Assert.AreEqual(1, eventArgs.UserProperties.Count); + Assert.AreEqual("test_name", eventArgs.UserProperties[0].Name); + Assert.AreEqual("test_value", eventArgs.UserProperties[0].Value); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs similarity index 93% rename from Source/MQTTnet.Tests/Client/MqttClient_Tests.cs rename to Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs index 8553ebd0..7b65e07f 100644 --- a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs @@ -21,7 +21,7 @@ using MQTTnet.Tests.Mockups; // ReSharper disable InconsistentNaming -namespace MQTTnet.Tests.Client +namespace MQTTnet.Tests.Clients.MqttClient { [TestClass] public sealed class MqttClient_Tests : BaseTestClass @@ -71,60 +71,34 @@ namespace MQTTnet.Tests.Client } [TestMethod] - [ExpectedException(typeof(MqttCommunicationException))] - public async Task Connect_To_Invalid_Server_Port_Not_Opened() + public async Task Connect_Disconnect_Connect() { - var client = new MqttFactory().CreateMqttClient(); - await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", 12345).WithTimeout(TimeSpan.FromSeconds(5)).Build()); - } - - [TestMethod] - [ExpectedException(typeof(OperationCanceledException))] - public async Task Connect_To_Invalid_Server_Wrong_IP() - { - var client = new MqttFactory().CreateMqttClient(); - using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2))) + using (var testEnvironment = CreateTestEnvironment()) { - await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build(), timeout.Token); + await testEnvironment.StartServer(); + + var clientOptions = testEnvironment.CreateDefaultClientOptions(); + var client = testEnvironment.CreateClient(); + + await client.ConnectAsync(clientOptions); + await client.DisconnectAsync(); + await client.ConnectAsync(clientOptions); } } - + [TestMethod] - [ExpectedException(typeof(MqttCommunicationException))] - public async Task Connect_To_Invalid_Server_Wrong_Protocol() + [ExpectedException(typeof(InvalidOperationException))] + public async Task Connect_Multiple_Times_Should_Fail() { - var client = new MqttFactory().CreateMqttClient(); - await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("http://127.0.0.1", 12345).WithTimeout(TimeSpan.FromSeconds(2)).Build()); - } - - [TestMethod] - public async Task ConnectTimeout_Throws_Exception() - { - var factory = new MqttFactory(); - using (var client = factory.CreateMqttClient()) + using (var testEnvironment = CreateTestEnvironment()) { - var disconnectHandlerCalled = false; - try - { - client.DisconnectedAsync += args => - { - disconnectHandlerCalled = true; - return CompletedTask.Instance; - }; + await testEnvironment.StartServer(); - await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build()); + var clientOptions = testEnvironment.CreateDefaultClientOptions(); + var client = testEnvironment.CreateClient(); - Assert.Fail("Must fail!"); - } - catch (Exception exception) - { - Assert.IsNotNull(exception); - Assert.IsInstanceOfType(exception, typeof(MqttCommunicationException)); - //Assert.IsInstanceOfType(exception.InnerException, typeof(SocketException)); - } - - await LongTestDelay(); // disconnected handler is called async - Assert.IsTrue(disconnectHandlerCalled); + await client.ConnectAsync(clientOptions); + await client.ConnectAsync(clientOptions); } } diff --git a/Source/MQTTnet.Tests/Client/MqttClientOptionsBuilder_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClientOptionsBuilder_Tests.cs similarity index 93% rename from Source/MQTTnet.Tests/Client/MqttClientOptionsBuilder_Tests.cs rename to Source/MQTTnet.Tests/Clients/MqttClientOptionsBuilder_Tests.cs index 59db1094..dac2404e 100644 --- a/Source/MQTTnet.Tests/Client/MqttClientOptionsBuilder_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/MqttClientOptionsBuilder_Tests.cs @@ -6,9 +6,8 @@ using System.Linq; using System.Text; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; -using MQTTnet.Extensions; -namespace MQTTnet.Tests.Client +namespace MQTTnet.Tests.Clients { [TestClass] public class MqttClientOptionsBuilder_Tests diff --git a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs index 44d692d9..9995ee82 100644 --- a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs +++ b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs @@ -238,6 +238,16 @@ namespace MQTTnet.Tests.Mockups } } + public MqttClientOptions CreateDefaultClientOptions() + { + return CreateDefaultClientOptionsBuilder().Build(); + } + + public MqttClientOptionsBuilder CreateDefaultClientOptionsBuilder() + { + return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort); + } + public ILowLevelMqttClient CreateLowLevelClient() { lock (_clients) diff --git a/Source/MQTTnet.Tests/Server/General.cs b/Source/MQTTnet.Tests/Server/General.cs index eeb75fc6..9d1b0c25 100644 --- a/Source/MQTTnet.Tests/Server/General.cs +++ b/Source/MQTTnet.Tests/Server/General.cs @@ -636,7 +636,7 @@ namespace MQTTnet.Tests.Server // nothing Assert.AreEqual(false, c1.IsConnected); - await c1.DisconnectAsync(); + await c1.TryDisconnectAsync(); Assert.AreEqual(false, c1.IsConnected); await Task.Delay(500); diff --git a/Source/MQTTnet/Client/Connecting/MqttClientConnectedEventArgs.cs b/Source/MQTTnet/Client/Connecting/MqttClientConnectedEventArgs.cs index efb28e83..2ea08978 100644 --- a/Source/MQTTnet/Client/Connecting/MqttClientConnectedEventArgs.cs +++ b/Source/MQTTnet/Client/Connecting/MqttClientConnectedEventArgs.cs @@ -15,7 +15,7 @@ namespace MQTTnet.Client /// /// Gets the authentication result. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// public MqttClientConnectResult ConnectResult { get; } } diff --git a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptions.cs b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptions.cs index 964e30a7..7f57ca2a 100644 --- a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptions.cs +++ b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptions.cs @@ -2,6 +2,9 @@ // 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; + namespace MQTTnet.Client { public sealed class MqttClientDisconnectOptions @@ -10,12 +13,24 @@ namespace MQTTnet.Client /// Gets or sets the reason code. /// MQTT 5.0.0+ feature. /// - public MqttClientDisconnectReason Reason { get; set; } = MqttClientDisconnectReason.NormalDisconnection; + public MqttClientDisconnectOptionsReason Reason { get; set; } = MqttClientDisconnectOptionsReason.NormalDisconnection; /// /// Gets or sets the reason string. /// MQTT 5.0.0+ feature. /// public string ReasonString { get; set; } + + /// + /// Gets or sets the session expiry interval. + /// MQTT 5.0.0+ feature. + /// + public uint SessionExpiryInterval { get; set; } + + /// + /// Gets or sets the user properties. + /// MQTT 5.0.0+ feature. + /// + public List UserProperties { get; set; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsBuilder.cs b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsBuilder.cs index 34f16b74..3c402309 100644 --- a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsBuilder.cs @@ -2,32 +2,62 @@ // 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; + namespace MQTTnet.Client { public sealed class MqttClientDisconnectOptionsBuilder { - MqttClientDisconnectReason _reason = MqttClientDisconnectReason.NormalDisconnection; + MqttClientDisconnectOptionsReason _reason = MqttClientDisconnectOptionsReason.NormalDisconnection; string _reasonString; - - public MqttClientDisconnectOptionsBuilder WithReasonString(string value) - { - _reasonString = value; - return this; - } - - public MqttClientDisconnectOptionsBuilder WithReason(MqttClientDisconnectReason value) - { - _reason = value; - return this; - } - + uint _sessionExpiryInterval; + List _userProperties; + public MqttClientDisconnectOptions Build() { return new MqttClientDisconnectOptions { Reason = _reason, - ReasonString = _reasonString + ReasonString = _reasonString, + UserProperties = _userProperties, + SessionExpiryInterval = _sessionExpiryInterval }; } + + public MqttClientDisconnectOptionsBuilder WithReason(MqttClientDisconnectOptionsReason value) + { + _reason = value; + return this; + } + + public MqttClientDisconnectOptionsBuilder WithReasonString(string value) + { + _reasonString = value; + return this; + } + + public MqttClientDisconnectOptionsBuilder WithSessionExpiryInterval(uint value) + { + _sessionExpiryInterval = value; + return this; + } + + public MqttClientDisconnectOptionsBuilder WithUserProperties(List userProperties) + { + _userProperties = userProperties; + return this; + } + + public MqttClientDisconnectOptionsBuilder WithUserProperty(string name, string value) + { + if (_userProperties == null) + { + _userProperties = new List(); + } + + _userProperties.Add(new MqttUserProperty(name, value)); + return this; + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsReason.cs b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsReason.cs new file mode 100644 index 00000000..c51ec995 --- /dev/null +++ b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsReason.cs @@ -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. + +namespace MQTTnet.Client +{ + /// + /// This enum only contains values which are valid when a client sends the reason to the server. + /// + public enum MqttClientDisconnectOptionsReason + { + NormalDisconnection = 0, + DisconnectWithWillMessage = 4, + UnspecifiedError = 128, + MalformedPacket = 129, + ProtocolError = 130, + ImplementationSpecificError = 131, + TopicNameInvalid = 144, + ReceiveMaximumExceeded = 147, + TopicAliasInvalid = 148, + PacketTooLarge = 149, + MessageRateTooHigh = 150, + QuotaExceeded = 151, + AdministrativeAction = 152, + PayloadFormatInvalid = 153 + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsValidator.cs b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsValidator.cs index b5fc4cc8..0a060483 100644 --- a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsValidator.cs +++ b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptionsValidator.cs @@ -28,7 +28,7 @@ namespace MQTTnet.Client Throw(nameof(options.ReasonString)); } - if (options.Reason != MqttClientDisconnectReason.NormalDisconnection) + if (options.Reason != MqttClientDisconnectOptionsReason.NormalDisconnection) { Throw(nameof(options.Reason)); } diff --git a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs index 009e0208..0a24e44f 100644 --- a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs +++ b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs @@ -3,27 +3,47 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections.Generic; +using MQTTnet.Packets; namespace MQTTnet.Client { public sealed class MqttClientDisconnectedEventArgs : EventArgs { - public bool ClientWasConnected { get; internal set; } + public MqttClientDisconnectedEventArgs( + bool clientWasConnected, + MqttClientConnectResult connectResult, + MqttClientDisconnectReason reason, + string reasonString, + List userProperties, + Exception exception) + { + ClientWasConnected = clientWasConnected; + ConnectResult = connectResult; + Exception = exception; + Reason = reason; + ReasonString = reasonString; + UserProperties = userProperties; + } - public Exception Exception { get; internal set; } + public bool ClientWasConnected { get; } /// - /// Gets the authentication result. - /// Hint: MQTT 5 feature only. + /// Gets the authentication result. + /// MQTT 5.0.0+ feature. /// - public MqttClientConnectResult ConnectResult { get; internal set; } + public MqttClientConnectResult ConnectResult { get; } + + public Exception Exception { get; } /// - /// Gets or sets the reason. - /// Hint: MQTT 5 feature only. + /// Gets or sets the reason. + /// MQTT 5.0.0+ feature. /// - public MqttClientDisconnectReason Reason { get; internal set; } - - public string ReasonString { get; internal set; } + public MqttClientDisconnectReason Reason { get; } + + public string ReasonString { get; } + + public List UserProperties { get; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index ed53a6c5..3ed4dd80 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -3,6 +3,7 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; @@ -36,13 +37,20 @@ namespace MQTTnet.Client readonly IMqttNetLogger _rootLogger; IMqttChannelAdapter _adapter; - CancellationTokenSource _backgroundCancellationTokenSource; + bool _cleanDisconnectInitiated; volatile int _connectionStatus; - MqttClientDisconnectReason _disconnectReason; + + // The value for this field can be set from two different enums. + // They contain the same values but the set is reduced in one case. + int _disconnectReason; string _disconnectReasonString; + List _disconnectUserProperties; + Task _keepAlivePacketsSenderTask; DateTime _lastPacketSentTimestamp; + + CancellationTokenSource _mqttClientAlive; MqttPacketDispatcher _packetDispatcher; Task _packetReceiverTask; AsyncQueue _publishPacketReceiverQueue; @@ -111,28 +119,29 @@ namespace MQTTnet.Client await _connectingEvent.InvokeAsync(new MqttClientConnectingEventArgs(options)); } + Cleanup(); + _packetIdentifierProvider.Reset(); _packetDispatcher = new MqttPacketDispatcher(); - _backgroundCancellationTokenSource = new CancellationTokenSource(); - var backgroundCancellationToken = _backgroundCancellationTokenSource.Token; + _mqttClientAlive = new CancellationTokenSource(); + var mqttClientAliveToken = _mqttClientAlive.Token; var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_inspectPacketEvent, _rootLogger), _rootLogger); _adapter = adapter; - using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken)) + if (cancellationToken.CanBeCanceled) { - _logger.Verbose("Trying to connect with server '{0}'.", options.ChannelOptions); - await adapter.ConnectAsync(effectiveCancellationToken.Token).ConfigureAwait(false); - _logger.Verbose("Connection with server established."); - - _publishPacketReceiverQueue?.Dispose(); - _publishPacketReceiverQueue = new AsyncQueue(); - _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken); - - _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken); - - connectResult = await AuthenticateAsync(options, effectiveCancellationToken.Token).ConfigureAwait(false); + connectResult = await ConnectInternal(cancellationToken).ConfigureAwait(false); + } + else + { + // Fall back to the general timeout specified in the options if the user passed + // CancellationToken.None or similar. + using (var timeout = new CancellationTokenSource(Options.Timeout)) + { + connectResult = await ConnectInternal(timeout.Token).ConfigureAwait(false); + } } _lastPacketSentTimestamp = DateTime.UtcNow; @@ -146,7 +155,7 @@ namespace MQTTnet.Client if (keepAliveInterval != TimeSpan.Zero) { - _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(backgroundCancellationToken), backgroundCancellationToken); + _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(mqttClientAliveToken), mqttClientAliveToken); } CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting); @@ -163,7 +172,7 @@ namespace MQTTnet.Client } catch (Exception exception) { - _disconnectReason = MqttClientDisconnectReason.UnspecifiedError; + _disconnectReason = (int)MqttClientDisconnectOptionsReason.UnspecifiedError; _logger.Error(exception, "Error while connecting with server."); @@ -183,26 +192,42 @@ namespace MQTTnet.Client ThrowIfDisposed(); var clientWasConnected = IsConnected; + if (DisconnectIsPendingOrFinished()) { return; } - + try { - _disconnectReason = MqttClientDisconnectReason.NormalDisconnection; + if (!clientWasConnected) + { + ThrowNotConnected(); + } + + _disconnectReason = (int)options.Reason; _cleanDisconnectInitiated = true; - if (clientWasConnected) + if (Options.ValidateFeatures) + { + MqttClientDisconnectOptionsValidator.ThrowIfNotSupported(options, _adapter.PacketFormatterAdapter.ProtocolVersion); + } + + // Sending the DISCONNECT may fail due to connection issues. The resulting exception + // must be throw to let the caller know that the disconnect is not a clean one. + var disconnectPacket = MqttPacketFactories.Disconnect.Create(options); + + if (cancellationToken.CanBeCanceled) { - if (Options.ValidateFeatures) - { - MqttClientDisconnectOptionsValidator.ThrowIfNotSupported(options, _adapter.PacketFormatterAdapter.ProtocolVersion); - } - - var disconnectPacket = MqttPacketFactories.Disconnect.Create(options); await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false); } + else + { + using (var timeout = new CancellationTokenSource(Options.Timeout)) + { + await SendAsync(disconnectPacket, timeout.Token).ConfigureAwait(false); + } + } } finally { @@ -307,7 +332,18 @@ namespace MQTTnet.Client var subscribePacket = MqttPacketFactories.Subscribe.Create(options); subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - var subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); + MqttSubAckPacket subAckPacket; + if (cancellationToken.CanBeCanceled) + { + subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); + } + else + { + using (var timeout = new CancellationTokenSource(Options.Timeout)) + { + subAckPacket = await SendAndReceiveAsync(subscribePacket, timeout.Token).ConfigureAwait(false); + } + } return _clientSubscribeResultFactory.Create(subscribePacket, subAckPacket); } @@ -335,7 +371,18 @@ namespace MQTTnet.Client var unsubscribePacket = MqttPacketFactories.Unsubscribe.Create(options); unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - var unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); + MqttUnsubAckPacket unsubAckPacket; + if (cancellationToken.CanBeCanceled) + { + unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); + } + else + { + using (var timeout = new CancellationTokenSource(Options.Timeout)) + { + unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, timeout.Token).ConfigureAwait(false); + } + } return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket); } @@ -412,19 +459,19 @@ namespace MQTTnet.Client { try { - _backgroundCancellationTokenSource?.Cancel(false); + _mqttClientAlive?.Cancel(false); } finally { - _backgroundCancellationTokenSource?.Dispose(); - _backgroundCancellationTokenSource = null; + _mqttClientAlive?.Dispose(); + _mqttClientAlive = null; _publishPacketReceiverQueue?.Dispose(); _publishPacketReceiverQueue = null; _adapter?.Dispose(); _adapter = null; - + _packetDispatcher?.Dispose(); _packetDispatcher = null; } @@ -435,6 +482,26 @@ namespace MQTTnet.Client return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand); } + async Task ConnectInternal(CancellationToken cancellationToken) + { + var backgroundCancellationToken = _mqttClientAlive.Token; + + using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken)) + { + _logger.Verbose("Trying to connect with server '{0}'.", Options.ChannelOptions); + await _adapter.ConnectAsync(effectiveCancellationToken.Token).ConfigureAwait(false); + _logger.Verbose("Connection with server established."); + + _publishPacketReceiverQueue?.Dispose(); + _publishPacketReceiverQueue = new AsyncQueue(); + + _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken); + _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken); + + return await AuthenticateAsync(Options, effectiveCancellationToken.Token).ConfigureAwait(false); + } + } + async Task DisconnectCoreAsync(Task sender, Exception exception, MqttClientConnectResult connectResult, bool clientWasConnected) { TryInitiateDisconnect(); @@ -461,10 +528,10 @@ namespace MQTTnet.Client try { _packetDispatcher.Dispose(new MqttClientDisconnectedException(exception)); - - var receiverTask = WaitForTaskAsync(_packetReceiverTask, sender); - var publishPacketReceiverTask = WaitForTaskAsync(_publishPacketReceiverTask, sender); - var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender); + + var receiverTask = _packetReceiverTask.WaitAsync(sender, _logger); + var publishPacketReceiverTask = _publishPacketReceiverTask.WaitAsync(sender, _logger); + var keepAliveTask = _keepAlivePacketsSenderTask.WaitAsync(sender, _logger); await Task.WhenAll(receiverTask, publishPacketReceiverTask, keepAliveTask).ConfigureAwait(false); } @@ -480,14 +547,13 @@ namespace MQTTnet.Client _logger.Info("Disconnected."); - var eventArgs = new MqttClientDisconnectedEventArgs - { - ClientWasConnected = clientWasConnected, - Exception = exception, - ConnectResult = connectResult, - Reason = _disconnectReason, - ReasonString = _disconnectReasonString - }; + var eventArgs = new MqttClientDisconnectedEventArgs( + clientWasConnected, + connectResult, + (MqttClientDisconnectReason)_disconnectReason, + _disconnectReasonString, + _disconnectUserProperties, + exception); // This handler must be executed in a new thread because otherwise a dead lock may happen // when trying to reconnect in that handler etc. @@ -550,7 +616,6 @@ namespace MQTTnet.Client { var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket); var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket); - await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); return eventArgs; @@ -569,8 +634,9 @@ namespace MQTTnet.Client Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) { - _disconnectReason = (MqttClientDisconnectReason)disconnectPacket.ReasonCode; + _disconnectReason = (int)disconnectPacket.ReasonCode; _disconnectReasonString = disconnectPacket.ReasonString; + _disconnectUserProperties = disconnectPacket.UserProperties; // Also dispatch disconnect to waiting threads to generate a proper exception. _packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket)); @@ -632,7 +698,6 @@ namespace MQTTnet.Client publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var pubAckPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - return _clientPublishResultFactory.Create(pubAckPacket); } @@ -708,7 +773,7 @@ namespace MQTTnet.Client { if (IsConnected) { - throw new MqttProtocolViolationException(message); + throw new InvalidOperationException(message); } } @@ -716,7 +781,7 @@ namespace MQTTnet.Client { if (!IsConnected) { - throw new MqttCommunicationException("The client is not connected."); + ThrowNotConnected(); } } @@ -738,13 +803,18 @@ namespace MQTTnet.Client } } + static void ThrowNotConnected() + { + throw new MqttCommunicationException("The client is not connected."); + } + void TryInitiateDisconnect() { lock (_disconnectLock) { try { - _backgroundCancellationTokenSource?.Cancel(false); + _mqttClientAlive?.Cancel(false); } catch (Exception exception) { @@ -864,7 +934,7 @@ namespace MQTTnet.Client { exception = aggregateException.GetBaseException(); } - + if (exception is OperationCanceledException) { } @@ -943,36 +1013,5 @@ namespace MQTTnet.Client _logger.Verbose("Stopped sending keep alive packets."); } } - - async Task WaitForTaskAsync(Task task, Task sender) - { - if (task == null) - { - return; - } - - if (task == sender) - { - // Return here to avoid deadlocks, but first any eventual exception in the task - // must be handled to avoid not getting an unhandled task exception - if (!task.IsFaulted) - { - return; - } - - // By accessing the Exception property the exception is considered handled and will - // not result in an unhandled task exception later by the finalizer - _logger.Warning(task.Exception, "Error while waiting for background task."); - return; - } - - try - { - await task.ConfigureAwait(false); - } - catch (OperationCanceledException) - { - } - } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index ba917aad..da2f2995 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -14,20 +14,28 @@ namespace MQTTnet.Client { public static class MqttClientExtensions { - public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null) + public static Task DisconnectAsync( + this IMqttClient client, + MqttClientDisconnectOptionsReason reason = MqttClientDisconnectOptionsReason.NormalDisconnection, + string reasonString = null, + uint sessionExpiryInterval = 0, + List userProperties = null, + CancellationToken cancellationToken = default) { if (client == null) { throw new ArgumentNullException(nameof(client)); } - return client.DisconnectAsync( - new MqttClientDisconnectOptions - { - Reason = reason, - ReasonString = reasonString - }, - CancellationToken.None); + var disconnectOptions = new MqttClientDisconnectOptions + { + Reason = reason, + ReasonString = reasonString, + SessionExpiryInterval = sessionExpiryInterval, + UserProperties = userProperties + }; + + return client.DisconnectAsync(disconnectOptions, cancellationToken); } public static Task PublishBinaryAsync( @@ -128,6 +136,29 @@ namespace MQTTnet.Client return mqttClient.SubscribeAsync(subscribeOptions, cancellationToken); } + public static async Task TryDisconnectAsync( + this IMqttClient client, + MqttClientDisconnectOptionsReason reason = MqttClientDisconnectOptionsReason.NormalDisconnection, + string reasonString = null) + { + if (client == null) + { + throw new ArgumentNullException(nameof(client)); + } + + try + { + await client.DisconnectAsync(reason, reasonString).ConfigureAwait(false); + return true; + } + catch + { + // Ignore all errors. + } + + return false; + } + public static async Task TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default) { if (client == null) diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index 96e48e39..029f5a08 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -339,16 +339,6 @@ namespace MQTTnet.Client public MqttClientOptionsBuilder WithUserProperty(string name, string value) { - if (name is null) - { - throw new ArgumentNullException(nameof(name)); - } - - if (value is null) - { - throw new ArgumentNullException(nameof(value)); - } - if (_options.UserProperties == null) { _options.UserProperties = new List(); diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptionsBuilder.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptionsBuilder.cs index c6b5890a..7075ef89 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptionsBuilder.cs @@ -14,26 +14,9 @@ namespace MQTTnet.Client { readonly MqttClientSubscribeOptions _subscribeOptions = new MqttClientSubscribeOptions(); - /// - /// Adds the user property to the subscribe options. - /// Hint: MQTT 5 feature only. - /// - /// The property name. - /// The property value. - /// A new instance of the class. - public MqttClientSubscribeOptionsBuilder WithUserProperty(string name, string value) + public MqttClientSubscribeOptions Build() { - if (name == null) throw new ArgumentNullException(nameof(name)); - if (value == null) throw new ArgumentNullException(nameof(value)); - - if (_subscribeOptions.UserProperties == null) - { - _subscribeOptions.UserProperties = new List(); - } - - _subscribeOptions.UserProperties.Add(new MqttUserProperty(name, value)); - - return this; + return _subscribeOptions; } public MqttClientSubscribeOptionsBuilder WithSubscriptionIdentifier(uint subscriptionIdentifier) @@ -42,7 +25,7 @@ namespace MQTTnet.Client { throw new MqttProtocolViolationException("Subscription identifier cannot be 0."); } - + _subscribeOptions.SubscriptionIdentifier = subscriptionIdentifier; return this; } @@ -54,19 +37,23 @@ namespace MQTTnet.Client bool retainAsPublished = false, MqttRetainHandling retainHandling = MqttRetainHandling.SendAtSubscribe) { - return WithTopicFilter(new MqttTopicFilter - { - Topic = topic, - QualityOfServiceLevel = qualityOfServiceLevel, - NoLocal = noLocal, - RetainAsPublished = retainAsPublished, - RetainHandling = retainHandling - }); + return WithTopicFilter( + new MqttTopicFilter + { + Topic = topic, + QualityOfServiceLevel = qualityOfServiceLevel, + NoLocal = noLocal, + RetainAsPublished = retainAsPublished, + RetainHandling = retainHandling + }); } public MqttClientSubscribeOptionsBuilder WithTopicFilter(Action topicFilterBuilder) { - if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder)); + if (topicFilterBuilder == null) + { + throw new ArgumentNullException(nameof(topicFilterBuilder)); + } var internalTopicFilterBuilder = new MqttTopicFilterBuilder(); topicFilterBuilder(internalTopicFilterBuilder); @@ -76,14 +63,20 @@ namespace MQTTnet.Client public MqttClientSubscribeOptionsBuilder WithTopicFilter(MqttTopicFilterBuilder topicFilterBuilder) { - if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder)); + if (topicFilterBuilder == null) + { + throw new ArgumentNullException(nameof(topicFilterBuilder)); + } return WithTopicFilter(topicFilterBuilder.Build()); } public MqttClientSubscribeOptionsBuilder WithTopicFilter(MqttTopicFilter topicFilter) { - if (topicFilter == null) throw new ArgumentNullException(nameof(topicFilter)); + if (topicFilter == null) + { + throw new ArgumentNullException(nameof(topicFilter)); + } if (_subscribeOptions.TopicFilters == null) { @@ -95,9 +88,20 @@ namespace MQTTnet.Client return this; } - public MqttClientSubscribeOptions Build() + /// + /// Adds the user property to the subscribe options. + /// MQTT 5.0.0+ feature. + /// + public MqttClientSubscribeOptionsBuilder WithUserProperty(string name, string value) { - return _subscribeOptions; + if (_subscribeOptions.UserProperties == null) + { + _subscribeOptions.UserProperties = new List(); + } + + _subscribeOptions.UserProperties.Add(new MqttUserProperty(name, value)); + + return this; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptionsBuilder.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptionsBuilder.cs index b9aabe4a..1db31117 100644 --- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptionsBuilder.cs @@ -8,48 +8,21 @@ using MQTTnet.Packets; namespace MQTTnet.Client { - public class MqttClientUnsubscribeOptionsBuilder + public sealed class MqttClientUnsubscribeOptionsBuilder { - private readonly MqttClientUnsubscribeOptions _unsubscribeOptions = new MqttClientUnsubscribeOptions(); + readonly MqttClientUnsubscribeOptions _unsubscribeOptions = new MqttClientUnsubscribeOptions(); - /// - /// Adds the user property to the unsubscribe options. - /// Hint: MQTT 5 feature only. - /// - /// The property name. - /// The property value. - /// A new instance of the class. - public MqttClientUnsubscribeOptionsBuilder WithUserProperty(string name, string value) + public MqttClientUnsubscribeOptions Build() { - if (name is null) throw new ArgumentNullException(nameof(name)); - if (value is null) throw new ArgumentNullException(nameof(value)); - - return WithUserProperty(new MqttUserProperty(name, value)); - } - - /// - /// Adds the user property to the unsubscribe options. - /// Hint: MQTT 5 feature only. - /// - /// The user property. - /// A new instance of the class. - public MqttClientUnsubscribeOptionsBuilder WithUserProperty(MqttUserProperty userProperty) - { - if (userProperty is null) throw new ArgumentNullException(nameof(userProperty)); - - if (_unsubscribeOptions.UserProperties is null) - { - _unsubscribeOptions.UserProperties = new List(); - } - - _unsubscribeOptions.UserProperties.Add(userProperty); - - return this; + return _unsubscribeOptions; } public MqttClientUnsubscribeOptionsBuilder WithTopicFilter(string topic) { - if (topic is null) throw new ArgumentNullException(nameof(topic)); + if (topic is null) + { + throw new ArgumentNullException(nameof(topic)); + } if (_unsubscribeOptions.TopicFilters is null) { @@ -63,14 +36,42 @@ namespace MQTTnet.Client public MqttClientUnsubscribeOptionsBuilder WithTopicFilter(MqttTopicFilter topicFilter) { - if (topicFilter is null) throw new ArgumentNullException(nameof(topicFilter)); + if (topicFilter is null) + { + throw new ArgumentNullException(nameof(topicFilter)); + } return WithTopicFilter(topicFilter.Topic); } - public MqttClientUnsubscribeOptions Build() + /// + /// Adds the user property to the unsubscribe options. + /// MQTT 5.0.0+ feature. + /// + public MqttClientUnsubscribeOptionsBuilder WithUserProperty(string name, string value) { - return _unsubscribeOptions; + return WithUserProperty(new MqttUserProperty(name, value)); + } + + /// + /// Adds the user property to the unsubscribe options. + /// MQTT 5.0.0+ feature. + /// + public MqttClientUnsubscribeOptionsBuilder WithUserProperty(MqttUserProperty userProperty) + { + if (userProperty is null) + { + throw new ArgumentNullException(nameof(userProperty)); + } + + if (_unsubscribeOptions.UserProperties is null) + { + _unsubscribeOptions.UserProperties = new List(); + } + + _unsubscribeOptions.UserProperties.Add(userProperty); + + return this; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs b/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs index 889bb977..1c72c039 100644 --- a/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs @@ -39,6 +39,8 @@ namespace MQTTnet.Formatter else { packet.ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason; + packet.UserProperties = clientDisconnectOptions.UserProperties; + packet.SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval; } return packet; diff --git a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs index f60bdab6..d7ae1f04 100644 --- a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs +++ b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs @@ -182,7 +182,6 @@ namespace MQTTnet.Implementations try { #if NET5_0_OR_GREATER - if (_networkStream != null) { await _networkStream.DisposeAsync().ConfigureAwait(false); diff --git a/Source/MQTTnet/Internal/TaskExtensions.cs b/Source/MQTTnet/Internal/TaskExtensions.cs index d4c8c4d7..a63b791a 100644 --- a/Source/MQTTnet/Internal/TaskExtensions.cs +++ b/Source/MQTTnet/Internal/TaskExtensions.cs @@ -2,8 +2,9 @@ // 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.Diagnostics; +using System; using System.Threading.Tasks; +using MQTTnet.Diagnostics; namespace MQTTnet.Internal { @@ -11,7 +12,8 @@ namespace MQTTnet.Internal { public static void RunInBackground(this Task task, MqttNetSourceLogger logger = null) { - task?.ContinueWith(t => + task?.ContinueWith( + t => { // Consume the exception first so that we get no exception regarding the not observed exception. var exception = t.Exception; @@ -19,5 +21,41 @@ namespace MQTTnet.Internal }, TaskContinuationOptions.OnlyOnFaulted); } + + public static async Task WaitAsync(this Task task, Task sender, MqttNetSourceLogger logger) + { + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + + if (task == null) + { + return; + } + + if (task == sender) + { + // Return here to avoid deadlocks, but first any eventual exception in the task + // must be handled to avoid not getting an unhandled task exception + if (!task.IsFaulted) + { + return; + } + + // By accessing the Exception property the exception is considered handled and will + // not result in an unhandled task exception later by the finalizer + logger.Warning(task.Exception, "Error while waiting for background task."); + return; + } + + try + { + await task.ConfigureAwait(false); + } + catch (OperationCanceledException) + { + } + } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/MqttApplicationMessageBuilder.cs b/Source/MQTTnet/MqttApplicationMessageBuilder.cs index 2f1674e3..16fbbaab 100644 --- a/Source/MQTTnet/MqttApplicationMessageBuilder.cs +++ b/Source/MQTTnet/MqttApplicationMessageBuilder.cs @@ -55,14 +55,8 @@ namespace MQTTnet /// /// Adds the content type to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithContentType(string contentType) { _contentType = contentType; @@ -71,18 +65,8 @@ namespace MQTTnet /// /// Adds the correlation data to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// The correlation data. - /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithCorrelationData(byte[] correlationData) { _correlationData = correlationData; @@ -91,18 +75,8 @@ namespace MQTTnet /// /// Adds the message expiry interval in seconds to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// The message expiry interval. - /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithMessageExpiryInterval(uint messageExpiryInterval) { _messageExpiryInterval = messageExpiryInterval; @@ -185,18 +159,8 @@ namespace MQTTnet /// /// Adds the payload format indicator to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// The payload format indicator. - /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithPayloadFormatIndicator(MqttPayloadFormatIndicator payloadFormatIndicator) { _payloadFormatIndicator = payloadFormatIndicator; @@ -220,18 +184,8 @@ namespace MQTTnet /// /// Adds the response topic to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// The response topic. - /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithResponseTopic(string responseTopic) { _responseTopic = responseTopic; @@ -251,18 +205,8 @@ namespace MQTTnet /// /// Adds the subscription identifier to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// The subscription identifier. - /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithSubscriptionIdentifier(uint subscriptionIdentifier) { if (_subscriptionIdentifiers == null) @@ -289,18 +233,8 @@ namespace MQTTnet /// /// Adds the topic alias to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// The topic alias. - /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithTopicAlias(ushort topicAlias) { _topicAlias = topicAlias; @@ -309,22 +243,8 @@ namespace MQTTnet /// /// Adds the user property to the message. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - /// - /// The property name. - /// - /// - /// The property value. - /// - /// - /// A new instance of the - /// - /// class. - /// public MqttApplicationMessageBuilder WithUserProperty(string name, string value) { if (_userProperties == null) diff --git a/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs b/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs index 20e2b2dc..bbc5d148 100644 --- a/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs @@ -4,17 +4,30 @@ using System; using System.Collections; +using System.Collections.Generic; +using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet.Server { public sealed class ClientDisconnectedEventArgs : EventArgs { - public ClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType, string endpoint, IDictionary sessionItems) + readonly MqttDisconnectPacket _disconnectPacket; + + public ClientDisconnectedEventArgs( + string clientId, + MqttDisconnectPacket disconnectPacket, + MqttClientDisconnectType disconnectType, + string endpoint, + IDictionary sessionItems) { ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); DisconnectType = disconnectType; Endpoint = endpoint; SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + + // The DISCONNECT packet can be null in case of a non clean disconnect or session takeover. + _disconnectPacket = disconnectPacket; } /// @@ -27,9 +40,37 @@ namespace MQTTnet.Server public string Endpoint { get; } + /// + /// Gets the reason code sent by the client. + /// Only available for clean disconnects. + /// MQTT 5.0.0+ feature. + /// + public MqttDisconnectReasonCode? ReasonCode => _disconnectPacket?.ReasonCode; + + /// + /// Gets the reason string sent by the client. + /// Only available for clean disconnects. + /// MQTT 5.0.0+ feature. + /// + public string ReasonString => _disconnectPacket?.ReasonString; + + /// + /// Gets the session expiry interval sent by the client. + /// Only available for clean disconnects. + /// MQTT 5.0.0+ feature. + /// + public uint SessionExpiryInterval => _disconnectPacket?.SessionExpiryInterval ?? 0; + /// /// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// public IDictionary SessionItems { get; } + + /// + /// Gets the user properties sent by the client. + /// Only available for clean disconnects. + /// MQTT 5.0.0+ feature. + /// + public List UserProperties => _disconnectPacket?.UserProperties; } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 4a09cd90..97895676 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -57,12 +57,12 @@ namespace MQTTnet.Server public IMqttChannelAdapter ChannelAdapter { get; } + public MqttDisconnectPacket DisconnectPacket { get; set; } + public string Endpoint { get; } public string Id => _connectPacket.ClientId; - public bool IsCleanDisconnect { get; private set; } - public bool IsRunning { get; private set; } public bool IsTakenOver { get; set; } @@ -95,8 +95,7 @@ namespace MQTTnet.Server { var cancellationToken = _cancellationToken.Token; - _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default) - .ConfigureAwait(false); + _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false); IsRunning = true; @@ -111,7 +110,9 @@ namespace MQTTnet.Server _cancellationToken = null; } - if (!IsTakenOver && !IsCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent) + var isCleanDisconnect = DisconnectPacket != null; + + if (!IsTakenOver && !isCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent) { var willPublishPacket = MqttPacketFactories.Publish.Create(Session.LatestConnectPacket); var willApplicationMessage = MqttApplicationMessageFactory.Create(willPublishPacket); @@ -431,9 +432,9 @@ namespace MQTTnet.Server { throw new MqttProtocolViolationException("A PINGRESP Packet is sent by the Server to the Client in response to a PINGREQ Packet only."); } - else if (currentPacket is MqttDisconnectPacket) + else if (currentPacket is MqttDisconnectPacket disconnectPacket) { - IsCleanDisconnect = true; + DisconnectPacket = disconnectPacket; return; } else diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index a21b6926..b76380ab 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -387,8 +387,8 @@ namespace MQTTnet.Server if (client.Id != null && !client.IsTakenOver && _eventContainer.ClientDisconnectedEvent.HasHandlers) { - var disconnectType = client.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean; - var eventArgs = new ClientDisconnectedEventArgs(client.Id, disconnectType, endpoint, client.Session.Items); + var disconnectType = client.DisconnectPacket != null ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean; + var eventArgs = new ClientDisconnectedEventArgs(client.Id, client.DisconnectPacket, disconnectType, endpoint, client.Session.Items); await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } @@ -592,8 +592,13 @@ namespace MQTTnet.Server if (_eventContainer.ClientDisconnectedEvent.HasHandlers) { - var eventArgs = new ClientDisconnectedEventArgs(oldClient.Id, MqttClientDisconnectType.Takeover, oldClient.Endpoint, oldClient.Session.Items); - + var eventArgs = new ClientDisconnectedEventArgs( + oldClient.Id, + null, + MqttClientDisconnectType.Takeover, + oldClient.Endpoint, + oldClient.Session.Items); + await _eventContainer.ClientDisconnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false); } }