From 0e815c626c3eb79e601b552c25968b04341c9ce3 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Mon, 1 May 2023 12:03:23 +0200 Subject: [PATCH 1/5] Add validation of string lengths (#1718) --- .github/workflows/ReleaseNotes.md | 1 + .../MqttPacketReaderWriterBenchmark.cs | 10 ++-- .../Formatter/MqttPacketWriter_Tests.cs | 16 ++++-- .../MQTTnet.Tests/MqttPacketWriter_Tests.cs | 4 +- Source/MQTTnet/Formatter/MqttBufferWriter.cs | 53 +++++++++++-------- .../Formatter/MqttPacketFormatterAdapter.cs | 13 +++-- .../Formatter/V3/MqttV3PacketFormatter.cs | 8 +-- .../Formatter/V5/MqttV5PacketEncoder.cs | 4 +- .../Formatter/V5/MqttV5PropertiesWriter.cs | 2 +- 9 files changed, 65 insertions(+), 46 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index e69de29b..1c12c6b7 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -0,0 +1 @@ +* [Core] Add validation of maximum string lengths (#1718). \ No newline at end of file diff --git a/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs b/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs index a0c474a1..0efc7ffa 100644 --- a/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs @@ -30,17 +30,17 @@ namespace MQTTnet.Benchmarks var writer = new MqttBufferWriter(4096, 65535); writer.WriteString("A relative short string."); - writer.WriteBinaryData(_demoPayload); + writer.WriteBinary(_demoPayload); writer.WriteByte(0x01); writer.WriteByte(0x02); writer.WriteVariableByteInteger(5647382); writer.WriteString("A relative short string."); writer.WriteVariableByteInteger(8574489); - writer.WriteBinaryData(_demoPayload); + writer.WriteBinary(_demoPayload); writer.WriteByte(2); writer.WriteByte(0x02); writer.WriteString("fjgffiogfhgfhoihgoireghreghreguhreguireoghreouighreouighreughreguiorehreuiohruiorehreuioghreug"); - writer.WriteBinaryData(_demoPayload); + writer.WriteBinary(_demoPayload); _readPayload = new ArraySegment(writer.GetBuffer(), 0, writer.Length).ToArray(); } @@ -83,11 +83,11 @@ namespace MQTTnet.Benchmarks writer.WriteVariableByteInteger(5647382); writer.WriteString("A relative short string."); writer.WriteVariableByteInteger(8574589); - writer.WriteBinaryData(_demoPayload); + writer.WriteBinary(_demoPayload); writer.WriteByte(2); writer.WriteByte(0x02); writer.WriteString("fjgffiogfhgfhoihgoireghreghreguhreguireoghreouighreouighreughreguiorehreuiohruiorehreuioghreug"); - writer.WriteBinaryData(_demoPayload); + writer.WriteBinary(_demoPayload); writer.Reset(0); } diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs index e2748f03..7a16ff30 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs @@ -1,6 +1,7 @@ using System; using System.Linq; using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Exceptions; using MQTTnet.Formatter; namespace MQTTnet.Tests.Formatter @@ -47,23 +48,32 @@ namespace MQTTnet.Tests.Formatter Assert.AreEqual(1234U, reader.ReadVariableByteInteger()); Assert.AreEqual(9876U, reader.ReadVariableByteInteger()); } + + [TestMethod] + [ExpectedException(typeof(MqttProtocolViolationException))] + public void Throw_If_String_Too_Long() + { + var writer = new MqttBufferWriter(4096, 65535); + + writer.WriteString(string.Empty.PadLeft(65536)); + } [TestMethod] public void Write_And_Read_Multiple_Times() { var writer = new MqttBufferWriter(4096, 65535); writer.WriteString("A relative short string."); - writer.WriteBinaryData(new byte[1234]); + writer.WriteBinary(new byte[1234]); writer.WriteByte(0x01); writer.WriteByte(0x02); writer.WriteVariableByteInteger(5647382); writer.WriteString("A relative short string."); writer.WriteVariableByteInteger(8574489); - writer.WriteBinaryData(new byte[48]); + writer.WriteBinary(new byte[48]); writer.WriteByte(2); writer.WriteByte(0x02); writer.WriteString("fjgffiogfhgfhoihgoireghreghreguhreguireoghreouighreouighreughreguiorehreuiohruiorehreuioghreug"); - writer.WriteBinaryData(new byte[3]); + writer.WriteBinary(new byte[3]); var readPayload = new ArraySegment(writer.GetBuffer(), 0, writer.Length).ToArray(); diff --git a/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs b/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs index 02eeb40a..0a3d3f5e 100644 --- a/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs +++ b/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs @@ -24,10 +24,10 @@ namespace MQTTnet.Tests writer.WriteString("1234567890"); Assert.AreEqual(10 + 2, writer.Length); - writer.WriteBinaryData(new byte[300]); + writer.WriteBinary(new byte[300]); Assert.AreEqual(300 + 2 + 12, writer.Length); - writer.WriteBinaryData(new byte[5000]); + writer.WriteBinary(new byte[5000]); Assert.AreEqual(5000 + 2 + 300 + 2 + 12, writer.Length); } } diff --git a/Source/MQTTnet/Formatter/MqttBufferWriter.cs b/Source/MQTTnet/Formatter/MqttBufferWriter.cs index 0849f5b3..14ad707b 100644 --- a/Source/MQTTnet/Formatter/MqttBufferWriter.cs +++ b/Source/MQTTnet/Formatter/MqttBufferWriter.cs @@ -19,7 +19,8 @@ namespace MQTTnet.Formatter /// public sealed class MqttBufferWriter { - public const uint VariableByteIntegerMaxValue = 268435455; + const uint VariableByteIntegerMaxValue = 268435455; + const int EncodedStringMaxLength = 65535; readonly int _maxBufferSize; @@ -112,25 +113,7 @@ namespace MQTTnet.Formatter WriteBinary(propertyWriter._buffer, 0, propertyWriter.Length); } - public void WriteBinary(byte[] buffer, int offset, int count) - { - if (buffer == null) - { - throw new ArgumentNullException(nameof(buffer)); - } - - if (count == 0) - { - return; - } - - EnsureAdditionalCapacity(count); - - MqttMemoryHelper.Copy(buffer, offset, _buffer, _position, count); - IncreasePosition(count); - } - - public void WriteBinaryData(byte[] value) + public void WriteBinary(byte[] value) { if (value == null || value.Length == 0) { @@ -155,6 +138,24 @@ namespace MQTTnet.Formatter } } + public void WriteBinary(byte[] buffer, int offset, int count) + { + if (buffer == null) + { + throw new ArgumentNullException(nameof(buffer)); + } + + if (count == 0) + { + return; + } + + EnsureAdditionalCapacity(count); + + MqttMemoryHelper.Copy(buffer, offset, _buffer, _position, count); + IncreasePosition(count); + } + public void WriteByte(byte @byte) { EnsureAdditionalCapacity(1); @@ -186,6 +187,13 @@ namespace MQTTnet.Formatter var writtenBytes = Encoding.UTF8.GetBytes(value, 0, value.Length, _buffer, _position + 2); + // From RFC: 1.5.4 UTF-8 Encoded String + // Unless stated otherwise all UTF-8 encoded strings can have any length in the range 0 to 65,535 bytes. + if (writtenBytes > EncodedStringMaxLength) + { + throw new MqttProtocolViolationException($"The maximum string length is 65535. The current string has a length of {writtenBytes}."); + } + _buffer[_position] = (byte)(writtenBytes >> 8); _buffer[_position + 1] = (byte)writtenBytes; @@ -273,8 +281,7 @@ namespace MQTTnet.Formatter newBufferLength *= 2; } - // Array.Resize will create a new array and copy the existing - // data to the new one. + // Array.Resize will create a new array and copy the existing data to the new one. Array.Resize(ref _buffer, newBufferLength); } @@ -285,6 +292,8 @@ namespace MQTTnet.Formatter if (_position > Length) { + // Also extend the position because we reached the end of the + // pre allocated buffer. Length = _position; } } diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index 580c59b3..af4e4976 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs @@ -31,10 +31,14 @@ namespace MQTTnet.Formatter public MqttProtocolVersion ProtocolVersion { get; private set; } = MqttProtocolVersion.Unknown; + public void Cleanup() + { + _bufferWriter.Cleanup(); + } + public MqttPacket Decode(ReceivedMqttPacket receivedMqttPacket) { ThrowIfFormatterNotSet(); - return _formatter.Decode(receivedMqttPacket); } @@ -50,11 +54,6 @@ namespace MQTTnet.Formatter return _formatter.Encode(packet); } - public void Cleanup() - { - _bufferWriter.Cleanup(); - } - public static IMqttPacketFormatter GetMqttPacketFormatter(MqttProtocolVersion protocolVersion, MqttBufferWriter bufferWriter) { if (protocolVersion == MqttProtocolVersion.Unknown) @@ -98,7 +97,7 @@ namespace MQTTnet.Formatter // Remove the mosquitto try_private flag (MQTT 3.1.1 Bridge). // This flag is accepted but not yet used. protocolLevel &= 0x7F; - + if (protocolName == "MQTT") { if (protocolLevel == 5) diff --git a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs index 92fbb836..808140b4 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs @@ -459,7 +459,7 @@ namespace MQTTnet.Formatter.V3 if (packet.WillFlag) { bufferWriter.WriteString(packet.WillTopic); - bufferWriter.WriteBinaryData(packet.WillMessage); + bufferWriter.WriteBinary(packet.WillMessage); } if (packet.Username != null) @@ -469,7 +469,7 @@ namespace MQTTnet.Formatter.V3 if (packet.Password != null) { - bufferWriter.WriteBinaryData(packet.Password); + bufferWriter.WriteBinary(packet.Password); } return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect); @@ -521,7 +521,7 @@ namespace MQTTnet.Formatter.V3 if (packet.WillFlag) { bufferWriter.WriteString(packet.WillTopic); - bufferWriter.WriteBinaryData(packet.WillMessage); + bufferWriter.WriteBinary(packet.WillMessage); } if (packet.Username != null) @@ -531,7 +531,7 @@ namespace MQTTnet.Formatter.V3 if (packet.Password != null) { - bufferWriter.WriteBinaryData(packet.Password); + bufferWriter.WriteBinary(packet.Password); } return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect); diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs index eaddaaad..8023c9c0 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs @@ -186,7 +186,7 @@ namespace MQTTnet.Formatter.V5 _propertiesWriter.Reset(); _bufferWriter.WriteString(packet.WillTopic); - _bufferWriter.WriteBinaryData(packet.WillMessage); + _bufferWriter.WriteBinary(packet.WillMessage); } if (packet.Username != null) @@ -196,7 +196,7 @@ namespace MQTTnet.Formatter.V5 if (packet.Password != null) { - _bufferWriter.WriteBinaryData(packet.Password); + _bufferWriter.WriteBinary(packet.Password); } return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect); diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs index 30da6de8..29c06566 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs @@ -335,7 +335,7 @@ namespace MQTTnet.Formatter.V5 } _bufferWriter.WriteByte((byte)id); - _bufferWriter.WriteBinaryData(value); + _bufferWriter.WriteBinary(value); } void WriteAsFourByteInteger(MqttPropertyId id, uint value) From 836ddf3bbcea3ed85b1d5b5feb6755df02e5da6c Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Mon, 1 May 2023 12:20:20 +0200 Subject: [PATCH 2/5] Update version --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0597dcdc..83f7c057 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -3,7 +3,7 @@ name: CI on: [push, pull_request] env: - VERSION: "4.2.0.${{github.run_number}}" + VERSION: "4.2.1.${{github.run_number}}" jobs: build: From b08d11c069e53a67075c0339322c9eb0f980fe3b Mon Sep 17 00:00:00 2001 From: henry Date: Mon, 1 May 2023 19:53:15 +0800 Subject: [PATCH 3/5] =?UTF-8?q?replace=20lock/semaphore=20by=20readerwrite?= =?UTF-8?q?rlockslim=20to=20avoid=20lock=20contenti=E2=80=A6=20(#1716)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * replace lock/semaphore by readerwriterlockslim to avoid lock contention while dispatch messages * Update MqttClientSubscriptionsManager.cs removed comment out codes * Update ReleaseNotes.md --------- Co-authored-by: Zeheng Feng Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com> --- .github/workflows/ReleaseNotes.md | 3 +- .../Internal/MqttClientSessionsManager.cs | 65 ++++++++++++++++--- .../MqttClientSubscriptionsManager.cs | 33 +++++++--- 3 files changed, 82 insertions(+), 19 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 1c12c6b7..5c093744 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1 +1,2 @@ -* [Core] Add validation of maximum string lengths (#1718). \ No newline at end of file +* [Core] Add validation of maximum string lengths (#1718). +* [Server] Improved performance by changing internal locking strategy for subscriptions (#1716, thanks to @zeheng). diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 627ffe6d..4053e438 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -36,7 +36,7 @@ namespace MQTTnet.Server // See the MqttSubscription object for a detailed explanation. readonly Dictionary _sessions = new Dictionary(4096); - readonly object _sessionsManagementLock = new object(); + readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim(); readonly HashSet _subscriberSessions = new HashSet(); public MqttClientSessionsManager( @@ -86,7 +86,8 @@ namespace MQTTnet.Server MqttSession session; - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterWriteLock(); + try { _sessions.TryGetValue(clientId, out session); _sessions.Remove(clientId); @@ -96,6 +97,10 @@ namespace MQTTnet.Server _subscriberSessions.Remove(session); } } + finally + { + _sessionsManagementLock.ExitWriteLock(); + } try { @@ -173,10 +178,15 @@ namespace MQTTnet.Server } List subscriberSessions; - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterReadLock(); + try { subscriberSessions = _subscriberSessions.ToList(); } + finally + { + _sessionsManagementLock.ExitReadLock(); + } // Calculate application message topic hash once for subscription checks MqttTopicHash.Calculate(applicationMessage.Topic, out var topicHash, out _, out _); @@ -256,13 +266,23 @@ namespace MQTTnet.Server { _createConnectionSyncRoot.Dispose(); - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterWriteLock(); + try { foreach (var sessionItem in _sessions) { sessionItem.Value.Dispose(); } } + finally + { + _sessionsManagementLock.ExitWriteLock(); + } + + if (_sessionsManagementLock != null) + { + _sessionsManagementLock.Dispose(); + } } public MqttClient GetClient(string id) @@ -310,7 +330,8 @@ namespace MQTTnet.Server { var result = new List(); - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterReadLock(); + try { foreach (var sessionItem in _sessions) { @@ -318,6 +339,10 @@ namespace MQTTnet.Server result.Add(sessionStatus); } } + finally + { + _sessionsManagementLock.ExitReadLock(); + } return Task.FromResult((IList)result); } @@ -415,7 +440,8 @@ namespace MQTTnet.Server public void OnSubscriptionsAdded(MqttSession clientSession, List topics) { - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterWriteLock(); + try { if (!clientSession.HasSubscribedTopics) { @@ -428,11 +454,16 @@ namespace MQTTnet.Server clientSession.AddSubscribedTopic(topic); } } + finally + { + _sessionsManagementLock.ExitWriteLock(); + } } public void OnSubscriptionsRemoved(MqttSession clientSession, List subscriptionTopics) { - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterWriteLock(); + try { foreach (var subscriptionTopic in subscriptionTopics) { @@ -445,6 +476,10 @@ namespace MQTTnet.Server _subscriberSessions.Remove(clientSession); } } + finally + { + _sessionsManagementLock.ExitWriteLock(); + } } public void Start() @@ -545,7 +580,8 @@ namespace MQTTnet.Server MqttSession oldSession; MqttClient oldClient; - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterWriteLock(); + try { MqttSession session; @@ -590,6 +626,10 @@ namespace MQTTnet.Server _clients[connectPacket.ClientId] = client; } + finally + { + _sessionsManagementLock.ExitWriteLock(); + } if (!connAckPacket.IsSessionPresent) { @@ -636,7 +676,8 @@ namespace MQTTnet.Server MqttSession GetClientSession(string clientId) { - lock (_sessionsManagementLock) + _sessionsManagementLock.EnterReadLock(); + try { if (!_sessions.TryGetValue(clientId, out var session)) { @@ -645,6 +686,10 @@ namespace MQTTnet.Server return session; } + finally + { + _sessionsManagementLock.ExitReadLock(); + } } async Task ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) @@ -695,4 +740,4 @@ namespace MQTTnet.Server return eventArgs; } } -} \ No newline at end of file +} diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index a54a0045..037d6489 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -31,7 +31,7 @@ namespace MQTTnet.Server readonly Dictionary _subscriptions = new Dictionary(); // Use subscription lock to maintain consistency across subscriptions and topic hash dictionaries - readonly AsyncLock _subscriptionsLock = new AsyncLock(); + readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim(); readonly Dictionary _wildcardSubscriptionsByTopicHash = new Dictionary(); public MqttClientSubscriptionsManager( @@ -51,7 +51,8 @@ namespace MQTTnet.Server var possibleSubscriptions = new List(); // Check for possible subscriptions. They might have collisions but this is fine. - using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult()) + _subscriptionsLock.EnterReadLock(); + try { if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var noWildcardSubscriptions)) { @@ -73,6 +74,10 @@ namespace MQTTnet.Server } } } + finally + { + _subscriptionsLock.ExitReadLock(); + } // The pre check has evaluated that nothing is subscribed. // If there were some possible candidates they get checked below @@ -157,7 +162,10 @@ namespace MQTTnet.Server public void Dispose() { - _subscriptionsLock.Dispose(); + if (_subscriptionsLock != null) + { + _subscriptionsLock.Dispose(); + } } public async Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) @@ -232,7 +240,8 @@ namespace MQTTnet.Server var removedSubscriptions = new List(); - using (await _subscriptionsLock.EnterAsync(cancellationToken).ConfigureAwait(false)) + _subscriptionsLock.EnterWriteLock(); + try { foreach (var topicFilter in unsubscribePacket.TopicFilters) { @@ -293,8 +302,11 @@ namespace MQTTnet.Server } } } - - _subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions); + finally + { + _subscriptionsLock.ExitWriteLock(); + _subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions); + } if (_eventContainer.ClientUnsubscribedTopicEvent.HasHandlers) { @@ -341,7 +353,8 @@ namespace MQTTnet.Server // Add to subscriptions and maintain topic hash dictionaries - using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult()) + _subscriptionsLock.EnterWriteLock(); + try { MqttTopicHash.Calculate(topicFilter.Topic, out var topicHash, out _, out var hasWildcard); @@ -391,6 +404,10 @@ namespace MQTTnet.Server subscriptions.Add(subscription); } } + finally + { + _subscriptionsLock.ExitWriteLock(); + } return new CreateSubscriptionResult { @@ -511,4 +528,4 @@ namespace MQTTnet.Server public MqttSubscription Subscription { get; set; } } } -} \ No newline at end of file +} From 99f4f46062611cd08e18b82515962b69e8c619e4 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Thu, 4 May 2023 22:19:06 +0200 Subject: [PATCH 4/5] Add new overloads for packet payload and will payload (#1721) --- .github/workflows/ReleaseNotes.md | 1 + .../Options/MqttClientOptionsBuilder.cs | 14 +++++- .../Publishing/MqttClientPublishResult.cs | 2 +- .../MQTTnet/MqttApplicationMessageBuilder.cs | 44 +++++++++---------- 4 files changed, 37 insertions(+), 24 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 5c093744..b8635cbe 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,2 +1,3 @@ * [Core] Add validation of maximum string lengths (#1718). +* [Client] Added overloads for setting packet payload and will payload (#1720). * [Server] Improved performance by changing internal locking strategy for subscriptions (#1716, thanks to @zeheng). diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index a7a0099f..3f9f3adb 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -1,4 +1,4 @@ -// Licensed to the .NET Foundation under one or more agreements. +// 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. @@ -421,6 +421,18 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithWillPayload(ArraySegment willPayload) + { + if (willPayload.Count == 0) + { + _options.WillPayload = null; + return this; + } + + _options.WillPayload = willPayload.ToArray(); + return this; + } + public MqttClientOptionsBuilder WithWillPayload(string willPayload) { if (string.IsNullOrEmpty(willPayload)) diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs index c9859d1b..e36ce75e 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Client /// Gets or sets the reason code. /// MQTT 5.0.0+ feature. /// - public MqttClientPublishReasonCode ReasonCode { get; } = MqttClientPublishReasonCode.Success; + public MqttClientPublishReasonCode ReasonCode { get; } /// /// Gets or sets the reason string. diff --git a/Source/MQTTnet/MqttApplicationMessageBuilder.cs b/Source/MQTTnet/MqttApplicationMessageBuilder.cs index a8a740a0..8d65bd31 100644 --- a/Source/MQTTnet/MqttApplicationMessageBuilder.cs +++ b/Source/MQTTnet/MqttApplicationMessageBuilder.cs @@ -2,16 +2,16 @@ // 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.Exceptions; -using MQTTnet.Internal; -using MQTTnet.Packets; -using MQTTnet.Protocol; using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Runtime.InteropServices; using System.Text; +using MQTTnet.Exceptions; +using MQTTnet.Internal; +using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet { @@ -20,9 +20,9 @@ namespace MQTTnet string _contentType; byte[] _correlationData; uint _messageExpiryInterval; - ArraySegment _payloadSegment; MqttPayloadFormatIndicator _payloadFormatIndicator; + ArraySegment _payloadSegment; MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce; string _responseTopic; bool _retain; @@ -89,13 +89,11 @@ namespace MQTTnet public MqttApplicationMessageBuilder WithPayload(byte[] payload) { - _payloadSegment = payload == null || payload.Length == 0 - ? EmptyBuffer.ArraySegment - : new ArraySegment(payload); + _payloadSegment = payload == null || payload.Length == 0 ? EmptyBuffer.ArraySegment : new ArraySegment(payload); return this; } - public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment payloadSegment) + public MqttApplicationMessageBuilder WithPayload(ArraySegment payloadSegment) { _payloadSegment = payloadSegment; return this; @@ -123,9 +121,7 @@ namespace MQTTnet public MqttApplicationMessageBuilder WithPayload(Stream payload) { - return payload == null - ? WithPayload(default(byte[])) - : WithPayload(payload, payload.Length - payload.Position); + return payload == null ? WithPayload(default(byte[])) : WithPayload(payload, payload.Length - payload.Position); } public MqttApplicationMessageBuilder WithPayload(Stream payload, long length) @@ -162,16 +158,6 @@ namespace MQTTnet return WithPayload(payloadBuffer); } - -#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1 - public MqttApplicationMessageBuilder WithPayloadSegment(ReadOnlyMemory payloadSegment) - { - return MemoryMarshal.TryGetArray(payloadSegment, out var segment) - ? WithPayloadSegment(segment) - : WithPayload(payloadSegment.ToArray()); - } -#endif - /// /// Adds the payload format indicator to the message. /// MQTT 5.0.0+ feature. @@ -182,6 +168,20 @@ namespace MQTTnet return this; } + public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment payloadSegment) + { + _payloadSegment = payloadSegment; + return this; + } + + +#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1 + public MqttApplicationMessageBuilder WithPayloadSegment(ReadOnlyMemory payloadSegment) + { + return MemoryMarshal.TryGetArray(payloadSegment, out var segment) ? WithPayloadSegment(segment) : WithPayload(payloadSegment.ToArray()); + } +#endif + /// /// The quality of service level. /// The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message From 449624fd72bb9962772bd45db18e3f2bbf73e089 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 6 May 2023 12:49:41 +0200 Subject: [PATCH 5/5] Expose proper connect result in clients Disconnected event (#1728) --- .github/workflows/ReleaseNotes.md | 1 + .../MqttFactoryExtensions.cs | 27 ++- .../ManagedMqttClient_Tests.cs | 32 +++ .../MqttClient/MqttClient_Connection_Tests.cs | 2 + .../MQTTnet.Tests/Mockups/TestEnvironment.cs | 2 +- .../Client/Internal/MqttClientEvents.cs | 18 ++ .../Internal/MqttClientResultFactory.cs | 13 ++ Source/MQTTnet/Client/MqttClient.cs | 214 ++++++++++-------- 8 files changed, 206 insertions(+), 103 deletions(-) create mode 100644 Source/MQTTnet/Client/Internal/MqttClientEvents.cs create mode 100644 Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index b8635cbe..0aafba3b 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,3 +1,4 @@ * [Core] Add validation of maximum string lengths (#1718). * [Client] Added overloads for setting packet payload and will payload (#1720). +* [Client] The proper connect result is now exposed in the _Disconnected_ event when authentication fails (#1139). * [Server] Improved performance by changing internal locking strategy for subscriptions (#1716, thanks to @zeheng). diff --git a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs index 3349ca5d..f5b9a733 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs @@ -12,22 +12,37 @@ namespace MQTTnet.Extensions.ManagedClient { public static IManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, IMqttClient mqttClient = null) { - if (factory == null) throw new ArgumentNullException(nameof(factory)); + if (factory == null) + { + throw new ArgumentNullException(nameof(factory)); + } if (mqttClient == null) { - return new ManagedMqttClient(factory.CreateMqttClient(), factory.DefaultLogger); + return new ManagedMqttClient(factory.CreateMqttClient(), factory.DefaultLogger); } - + return new ManagedMqttClient(mqttClient, factory.DefaultLogger); } - + public static IManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, IMqttNetLogger logger) { - if (factory == null) throw new ArgumentNullException(nameof(factory)); - if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (factory == null) + { + throw new ArgumentNullException(nameof(factory)); + } + + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } return new ManagedMqttClient(factory.CreateMqttClient(logger), logger); } + + public static ManagedMqttClientOptionsBuilder CreateManagedMqttClientOptionsBuilder(this MqttFactory _) + { + return new ManagedMqttClientOptionsBuilder(); + } } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs index f16d55fa..66551983 100644 --- a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs @@ -22,6 +22,38 @@ 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() { diff --git a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs index 1cdcd760..4a2e1f06 100644 --- a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs @@ -127,6 +127,8 @@ namespace MQTTnet.Tests.Clients.MqttClient // Perform a clean disconnect. await client.DisconnectAsync(disconnectOptions); + await LongTestDelay(); + Assert.IsNotNull(eventArgs); Assert.AreEqual(MqttDisconnectReasonCode.MessageRateTooHigh, eventArgs.ReasonCode); } diff --git a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs index 9995ee82..b3543706 100644 --- a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs +++ b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs @@ -245,7 +245,7 @@ namespace MQTTnet.Tests.Mockups public MqttClientOptionsBuilder CreateDefaultClientOptionsBuilder() { - return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort); + return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort).WithClientId(TestContext.TestName + "_" + Guid.NewGuid()); } public ILowLevelMqttClient CreateLowLevelClient() diff --git a/Source/MQTTnet/Client/Internal/MqttClientEvents.cs b/Source/MQTTnet/Client/Internal/MqttClientEvents.cs new file mode 100644 index 00000000..fe786e82 --- /dev/null +++ b/Source/MQTTnet/Client/Internal/MqttClientEvents.cs @@ -0,0 +1,18 @@ +// 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.Diagnostics; +using MQTTnet.Internal; + +namespace MQTTnet.Client.Internal +{ + public sealed class MqttClientEvents + { + public AsyncEvent ApplicationMessageReceivedEvent { get; } = new AsyncEvent(); + public AsyncEvent ConnectedEvent { get; } = new AsyncEvent(); + public AsyncEvent ConnectingEvent { get; } = new AsyncEvent(); + public AsyncEvent DisconnectedEvent { get; } = new AsyncEvent(); + public AsyncEvent InspectPacketEvent { get; } = new AsyncEvent(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs b/Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs new file mode 100644 index 00000000..ef2800fe --- /dev/null +++ b/Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs @@ -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.Client.Internal +{ + public static class MqttClientResultFactory + { + public static readonly MqttClientPublishResultFactory PublishResult = new MqttClientPublishResultFactory(); + public static readonly MqttClientSubscribeResultFactory SubscribeResult = new MqttClientSubscribeResultFactory(); + public static readonly MqttClientUnsubscribeResultFactory UnsubscribeResult = new MqttClientUnsubscribeResultFactory(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 876c1390..0719ed1f 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -7,6 +7,7 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; +using MQTTnet.Client.Internal; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Formatter; @@ -20,17 +21,9 @@ namespace MQTTnet.Client public sealed class MqttClient : Disposable, IMqttClient { readonly IMqttClientAdapterFactory _adapterFactory; - readonly AsyncEvent _applicationMessageReceivedEvent = new AsyncEvent(); - readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory(); - readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory(); - readonly MqttClientUnsubscribeResultFactory _clientUnsubscribeResultFactory = new MqttClientUnsubscribeResultFactory(); - - readonly AsyncEvent _connectedEvent = new AsyncEvent(); - readonly AsyncEvent _connectingEvent = new AsyncEvent(); - readonly AsyncEvent _disconnectedEvent = new AsyncEvent(); readonly object _disconnectLock = new object(); - readonly AsyncEvent _inspectPacketEvent = new AsyncEvent(); + readonly MqttClientEvents _events = new MqttClientEvents(); readonly MqttNetSourceLogger _logger; readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); @@ -65,32 +58,32 @@ namespace MQTTnet.Client public event Func ApplicationMessageReceivedAsync { - add => _applicationMessageReceivedEvent.AddHandler(value); - remove => _applicationMessageReceivedEvent.RemoveHandler(value); + add => _events.ApplicationMessageReceivedEvent.AddHandler(value); + remove => _events.ApplicationMessageReceivedEvent.RemoveHandler(value); } public event Func ConnectedAsync { - add => _connectedEvent.AddHandler(value); - remove => _connectedEvent.RemoveHandler(value); + add => _events.ConnectedEvent.AddHandler(value); + remove => _events.ConnectedEvent.RemoveHandler(value); } public event Func ConnectingAsync { - add => _connectingEvent.AddHandler(value); - remove => _connectingEvent.RemoveHandler(value); + add => _events.ConnectingEvent.AddHandler(value); + remove => _events.ConnectingEvent.RemoveHandler(value); } public event Func DisconnectedAsync { - add => _disconnectedEvent.AddHandler(value); - remove => _disconnectedEvent.RemoveHandler(value); + add => _events.DisconnectedEvent.AddHandler(value); + remove => _events.DisconnectedEvent.RemoveHandler(value); } public event Func InspectPacketAsync { - add => _inspectPacketEvent.AddHandler(value); - remove => _inspectPacketEvent.RemoveHandler(value); + add => _events.InspectPacketEvent.AddHandler(value); + remove => _events.InspectPacketEvent.RemoveHandler(value); } public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected; @@ -114,9 +107,9 @@ namespace MQTTnet.Client { Options = options; - if (_connectingEvent.HasHandlers) + if (_events.ConnectingEvent.HasHandlers) { - await _connectingEvent.InvokeAsync(new MqttClientConnectingEventArgs(options)); + await _events.ConnectingEvent.InvokeAsync(new MqttClientConnectingEventArgs(options)); } Cleanup(); @@ -127,7 +120,7 @@ namespace MQTTnet.Client _mqttClientAlive = new CancellationTokenSource(); var mqttClientAliveToken = _mqttClientAlive.Token; - var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_inspectPacketEvent, _rootLogger), _rootLogger); + var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_events.InspectPacketEvent, _rootLogger), _rootLogger); _adapter = adapter; if (cancellationToken.CanBeCanceled) @@ -149,34 +142,35 @@ namespace MQTTnet.Client var keepAliveInterval = Options.KeepAlivePeriod; if (connectResult.ServerKeepAlive > 0) { - _logger.Info($"Using keep alive value ({connectResult.ServerKeepAlive}) sent from the server."); + _logger.Info($"Using keep alive value ({connectResult.ServerKeepAlive}) sent from the server"); keepAliveInterval = TimeSpan.FromSeconds(connectResult.ServerKeepAlive); } if (keepAliveInterval != TimeSpan.Zero) { - _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(mqttClientAliveToken), mqttClientAliveToken); + _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessages(mqttClientAliveToken), mqttClientAliveToken); } CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting); - _logger.Info("Connected."); + _logger.Info("Connected"); - if (_connectedEvent.HasHandlers) - { - var eventArgs = new MqttClientConnectedEventArgs(connectResult); - await _connectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); - } + await OnConnected(connectResult).ConfigureAwait(false); return connectResult; } catch (Exception exception) { + if (exception is MqttConnectingFailedException connectingFailedException) + { + connectResult = connectingFailedException.Result; + } + _disconnectReason = (int)MqttClientDisconnectOptionsReason.UnspecifiedError; - _logger.Error(exception, "Error while connecting with server."); + _logger.Error(exception, "Error while connecting with server"); - await DisconnectInternalAsync(null, exception, connectResult).ConfigureAwait(false); + await DisconnectInternal(null, exception, connectResult).ConfigureAwait(false); throw; } @@ -219,19 +213,19 @@ namespace MQTTnet.Client if (cancellationToken.CanBeCanceled) { - await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false); + await Send(disconnectPacket, cancellationToken).ConfigureAwait(false); } else { using (var timeout = new CancellationTokenSource(Options.Timeout)) { - await SendAsync(disconnectPacket, timeout.Token).ConfigureAwait(false); + await Send(disconnectPacket, timeout.Token).ConfigureAwait(false); } } } finally { - await DisconnectCoreAsync(null, null, null, clientWasConnected).ConfigureAwait(false); + await DisconnectCore(null, null, null, clientWasConnected).ConfigureAwait(false); } } @@ -239,13 +233,13 @@ namespace MQTTnet.Client { if (cancellationToken.CanBeCanceled) { - await SendAndReceiveAsync(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false); + await Request(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false); } else { using (var timeout = new CancellationTokenSource(Options.Timeout)) { - await SendAndReceiveAsync(MqttPingReqPacket.Instance, timeout.Token).ConfigureAwait(false); + await Request(MqttPingReqPacket.Instance, timeout.Token).ConfigureAwait(false); } } } @@ -274,11 +268,11 @@ namespace MQTTnet.Client } case MqttQualityOfServiceLevel.AtLeastOnce: { - return PublishAtLeastOnceAsync(publishPacket, cancellationToken); + return PublishAtLeastOnce(publishPacket, cancellationToken); } case MqttQualityOfServiceLevel.ExactlyOnce: { - return PublishExactlyOnceAsync(publishPacket, cancellationToken); + return PublishExactlyOnce(publishPacket, cancellationToken); } default: { @@ -306,7 +300,7 @@ namespace MQTTnet.Client UserProperties = data.UserProperties }; - return SendAsync(authPacket, cancellationToken); + return Send(authPacket, cancellationToken); } public async Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default) @@ -335,17 +329,17 @@ namespace MQTTnet.Client MqttSubAckPacket subAckPacket; if (cancellationToken.CanBeCanceled) { - subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); + subAckPacket = await Request(subscribePacket, cancellationToken).ConfigureAwait(false); } else { using (var timeout = new CancellationTokenSource(Options.Timeout)) { - subAckPacket = await SendAndReceiveAsync(subscribePacket, timeout.Token).ConfigureAwait(false); + subAckPacket = await Request(subscribePacket, timeout.Token).ConfigureAwait(false); } } - return _clientSubscribeResultFactory.Create(subscribePacket, subAckPacket); + return MqttClientResultFactory.SubscribeResult.Create(subscribePacket, subAckPacket); } public async Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default) @@ -374,17 +368,17 @@ namespace MQTTnet.Client MqttUnsubAckPacket unsubAckPacket; if (cancellationToken.CanBeCanceled) { - unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); + unsubAckPacket = await Request(unsubscribePacket, cancellationToken).ConfigureAwait(false); } else { using (var timeout = new CancellationTokenSource(Options.Timeout)) { - unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, timeout.Token).ConfigureAwait(false); + unsubAckPacket = await Request(unsubscribePacket, timeout.Token).ConfigureAwait(false); } } - return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket); + return MqttClientResultFactory.UnsubscribeResult.Create(unsubscribePacket, unsubAckPacket); } protected override void Dispose(bool disposing) @@ -408,7 +402,7 @@ namespace MQTTnet.Client if (!eventArgs.ProcessingFailed) { var pubAckPacket = MqttPacketFactories.PubAck.Create(eventArgs); - return SendAsync(pubAckPacket, cancellationToken); + return Send(pubAckPacket, cancellationToken); } } else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) @@ -416,7 +410,7 @@ namespace MQTTnet.Client if (!eventArgs.ProcessingFailed) { var pubRecPacket = MqttPacketFactories.PubRec.Create(eventArgs); - return SendAsync(pubRecPacket, cancellationToken); + return Send(pubRecPacket, cancellationToken); } } else @@ -427,18 +421,26 @@ namespace MQTTnet.Client return CompletedTask.Instance; } - async Task AuthenticateAsync(MqttClientOptions options, CancellationToken cancellationToken) + async Task Authenticate(MqttClientOptions options, CancellationToken cancellationToken) { MqttClientConnectResult result; try { var connectPacket = MqttPacketFactories.Connect.Create(options); + await Send(connectPacket, cancellationToken).ConfigureAwait(false); - var connAckPacket = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false); + var receivedPacket = await Receive(cancellationToken).ConfigureAwait(false); - var clientConnectResultFactory = new MqttClientConnectResultFactory(); - result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion); + if (receivedPacket is MqttConnAckPacket connAckPacket) + { + var clientConnectResultFactory = new MqttClientConnectResultFactory(); + result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion); + } + else + { + throw new NotSupportedException("Extended authentication handler is not yet supported"); + } } catch (Exception exception) { @@ -495,14 +497,16 @@ namespace MQTTnet.Client _publishPacketReceiverQueue?.Dispose(); _publishPacketReceiverQueue = new AsyncQueue(); - _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken); - _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken); + var connectResult = await Authenticate(Options, effectiveCancellationToken.Token).ConfigureAwait(false); - return await AuthenticateAsync(Options, effectiveCancellationToken.Token).ConfigureAwait(false); + _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken); + _packetReceiverTask = Task.Run(() => ReceivePacketsLoop(backgroundCancellationToken), backgroundCancellationToken); + + return connectResult; } } - async Task DisconnectCoreAsync(Task sender, Exception exception, MqttClientConnectResult connectResult, bool clientWasConnected) + async Task DisconnectCore(Task sender, Exception exception, MqttClientConnectResult connectResult, bool clientWasConnected) { TryInitiateDisconnect(); @@ -557,17 +561,17 @@ namespace MQTTnet.Client // This handler must be executed in a new thread because otherwise a dead lock may happen // when trying to reconnect in that handler etc. - Task.Run(() => _disconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger); + Task.Run(() => _events.DisconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger); } } - Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult) + Task DisconnectInternal(Task sender, Exception exception, MqttClientConnectResult connectResult) { var clientWasConnected = IsConnected; if (!DisconnectIsPendingOrFinished()) { - return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected); + return DisconnectCore(sender, exception, connectResult, clientWasConnected); } return CompletedTask.Instance; @@ -612,15 +616,26 @@ namespace MQTTnet.Client } } - async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) + async Task HandleReceivedApplicationMessage(MqttPublishPacket publishPacket) { var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket); var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket); - await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + await _events.ApplicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); return eventArgs; } + Task OnConnected(MqttClientConnectResult connectResult) + { + if (_events.ConnectedEvent.HasHandlers) + { + var eventArgs = new MqttClientConnectedEventArgs(connectResult); + return _events.ConnectedEvent.InvokeAsync(eventArgs); + } + + return CompletedTask.Instance; + } + Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket) { var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler; @@ -641,7 +656,7 @@ namespace MQTTnet.Client // Also dispatch disconnect to waiting threads to generate a proper exception. _packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket)); - return DisconnectInternalAsync(_packetReceiverTask, null, null); + return DisconnectInternal(_packetReceiverTask, null, null); } async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken) @@ -657,7 +672,7 @@ namespace MQTTnet.Client } var publishPacket = publishPacketDequeueResult.Item; - var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); + var eventArgs = await HandleReceivedApplicationMessage(publishPacket).ConfigureAwait(false); if (eventArgs.AutoAcknowledge) { @@ -681,7 +696,7 @@ namespace MQTTnet.Client // The packet is unknown. Probably due to a restart of the client. // So wen send this to the server to trigger a full resend of the message. var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); - return SendAsync(pubRelPacket, cancellationToken); + return Send(pubRelPacket, cancellationToken); } return CompletedTask.Instance; @@ -690,39 +705,56 @@ namespace MQTTnet.Client Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { var pubCompPacket = MqttPacketFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); - return SendAsync(pubCompPacket, cancellationToken); + return Send(pubCompPacket, cancellationToken); } - async Task PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + async Task PublishAtLeastOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - var pubAckPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - return _clientPublishResultFactory.Create(pubAckPacket); + var pubAckPacket = await Request(publishPacket, cancellationToken).ConfigureAwait(false); + return MqttClientResultFactory.PublishResult.Create(pubAckPacket); } async Task PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false); + await Send(publishPacket, cancellationToken).ConfigureAwait(false); - return _clientPublishResultFactory.Create(null); + return MqttClientResultFactory.PublishResult.Create(null); } - async Task PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + async Task PublishExactlyOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); + var pubRecPacket = await Request(publishPacket, cancellationToken).ConfigureAwait(false); var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); + var pubCompPacket = await Request(pubRelPacket, cancellationToken).ConfigureAwait(false); - return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket); + return MqttClientResultFactory.PublishResult.Create(pubRecPacket, pubCompPacket); } - async Task SendAndReceiveAsync(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket + async Task Receive(CancellationToken cancellationToken) + { + var packetTask = _adapter.ReceivePacketAsync(cancellationToken); + + MqttPacket packet; + if (packetTask.IsCompleted) + { + packet = packetTask.Result; + } + else + { + packet = await packetTask.ConfigureAwait(false); + } + + return packet; + } + + async Task Request(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket { cancellationToken.ThrowIfCancellationRequested(); @@ -736,7 +768,7 @@ namespace MQTTnet.Client { try { - await SendAsync(requestPacket, cancellationToken).ConfigureAwait(false); + await Send(requestPacket, cancellationToken).ConfigureAwait(false); } catch (Exception exception) { @@ -760,7 +792,7 @@ namespace MQTTnet.Client } } - Task SendAsync(MqttPacket packet, CancellationToken cancellationToken) + Task Send(MqttPacket packet, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); @@ -823,7 +855,7 @@ namespace MQTTnet.Client } } - async Task TryProcessReceivedPacketAsync(MqttPacket packet, CancellationToken cancellationToken) + async Task TryProcessReceivedPacket(MqttPacket packet, CancellationToken cancellationToken) { try { @@ -884,11 +916,11 @@ namespace MQTTnet.Client _packetDispatcher.FailAll(exception); - await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false); + await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false); } } - async Task TryReceivePacketsAsync(CancellationToken cancellationToken) + async Task ReceivePacketsLoop(CancellationToken cancellationToken) { try { @@ -896,17 +928,7 @@ namespace MQTTnet.Client while (!cancellationToken.IsCancellationRequested) { - MqttPacket packet; - var packetTask = _adapter.ReceivePacketAsync(cancellationToken); - - if (packetTask.IsCompleted) - { - packet = packetTask.Result; - } - else - { - packet = await packetTask.ConfigureAwait(false); - } + var packet = await Receive(cancellationToken).ConfigureAwait(false); if (cancellationToken.IsCancellationRequested) { @@ -915,12 +937,12 @@ namespace MQTTnet.Client if (packet == null) { - await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false); + await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false); return; } - await TryProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); + await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false); } } catch (Exception exception) @@ -949,7 +971,7 @@ namespace MQTTnet.Client _packetDispatcher.FailAll(exception); - await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false); + await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false); } finally { @@ -957,7 +979,7 @@ namespace MQTTnet.Client } } - async Task TrySendKeepAliveMessagesAsync(CancellationToken cancellationToken) + async Task TrySendKeepAliveMessages(CancellationToken cancellationToken) { try { @@ -1006,7 +1028,7 @@ namespace MQTTnet.Client _logger.Error(exception, "Error exception while sending/receiving keep alive packets."); } - await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false); + await DisconnectInternal(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false); } finally {