From b98417ab391cb68f8d815fb11087885e513e2f3f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?G=C3=BCnther=20Foidl?= Date: Thu, 19 Jan 2023 20:08:51 +0100 Subject: [PATCH 1/2] Avoid some full memory barriers (#1590) * Avoid some full memory barriers * Reset in stats itself * Updated MqttClientStatistics * Fixed order of members as artifact from merge-conflict --- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 29 +++++++---- Source/MQTTnet/Internal/AsyncQueue.cs | 14 +++++ Source/MQTTnet/MQTTnet.csproj | 2 +- .../Server/Internal/MqttClientStatistics.cs | 51 +++++++++++-------- 4 files changed, 63 insertions(+), 33 deletions(-) diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 8e08dd0f..47e28b31 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -33,8 +33,7 @@ namespace MQTTnet.Adapter readonly AsyncLock _syncRoot = new AsyncLock(); - long _bytesReceived; - long _bytesSent; + Statistics _statistics; // mutable struct, don't make readonly! public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, MqttPacketInspector packetInspector, IMqttNetLogger logger) { @@ -51,9 +50,9 @@ namespace MQTTnet.Adapter _logger = logger.WithSource(nameof(MqttChannelAdapter)); } - public long BytesReceived => Interlocked.Read(ref _bytesReceived); + public long BytesReceived => Volatile.Read(ref _statistics._bytesReceived); - public long BytesSent => Interlocked.Read(ref _bytesSent); + public long BytesSent => Volatile.Read(ref _statistics._bytesSent); public X509Certificate2 ClientCertificate => _channel.ClientCertificate; @@ -147,7 +146,7 @@ namespace MQTTnet.Adapter _packetInspector?.EndReceivePacket(); - Interlocked.Add(ref _bytesSent, receivedPacket.TotalLength); + Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength); if (PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.Unknown) { @@ -181,11 +180,7 @@ namespace MQTTnet.Adapter return null; } - public void ResetStatistics() - { - Interlocked.Exchange(ref _bytesReceived, 0L); - Interlocked.Exchange(ref _bytesSent, 0L); - } + public void ResetStatistics() => _statistics.Reset(); public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken) { @@ -216,7 +211,7 @@ namespace MQTTnet.Adapter await _channel.WriteAsync(packetBuffer.Packet, true, cancellationToken).ConfigureAwait(false); } - Interlocked.Add(ref _bytesReceived, packetBuffer.Length); + Interlocked.Add(ref _statistics._bytesReceived, packetBuffer.Length); } catch (Exception exception) { @@ -459,5 +454,17 @@ namespace MQTTnet.Adapter throw new MqttCommunicationException(exception); } + + private struct Statistics + { + public long _bytesReceived; + public long _bytesSent; + + public void Reset() + { + Volatile.Write(ref _bytesReceived, 0); + Volatile.Write(ref _bytesSent, 0); + } + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Internal/AsyncQueue.cs b/Source/MQTTnet/Internal/AsyncQueue.cs index 52381a32..22e12b38 100644 --- a/Source/MQTTnet/Internal/AsyncQueue.cs +++ b/Source/MQTTnet/Internal/AsyncQueue.cs @@ -22,7 +22,11 @@ namespace MQTTnet.Internal public void Clear() { +#if NETCOREAPP3_1_OR_GREATER + _queue.Clear(); +#else Interlocked.Exchange(ref _queue, new ConcurrentQueue()); +#endif } public void Dispose() @@ -32,6 +36,16 @@ namespace MQTTnet.Internal _signal.Dispose(); _isDisposed = true; + +#if !NETSTANDARD1_3 + if (typeof(IDisposable).IsAssignableFrom(typeof(TItem))) + { + while (_queue.TryDequeue(out TItem item)) + { + (item as IDisposable).Dispose(); + } + } +#endif } } diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 12044cfe..6cbeb03d 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -89,4 +89,4 @@ - + \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs b/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs index 28f8e9cf..8fd96426 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs @@ -10,14 +10,14 @@ namespace MQTTnet.Server { public sealed class MqttClientStatistics { - long _receivedApplicationMessagesCount; + Statistics _statistics = new Statistics // mutable struct, don't make readonly! + { + // Start with 1 because the CONNACK packet is not counted here. + _receivedPacketsCount = 1, - // Start with 1 because the CONNACK packet is not counted here. - long _receivedPacketsCount = 1; - long _sentApplicationMessagesCount; - - // Start with 1 because the CONNECT packet is not counted here. - long _sentPacketsCount = 1; + // Start with 1 because the CONNECT packet is not counted here. + _sentPacketsCount = 1 + }; public MqttClientStatistics() { @@ -30,7 +30,6 @@ namespace MQTTnet.Server } public DateTime ConnectedTimestamp { get; } - public DateTime LastNonKeepAlivePacketReceivedTimestamp { get; private set; } /// @@ -43,13 +42,13 @@ namespace MQTTnet.Server /// public DateTime LastPacketSentTimestamp { get; private set; } - public long ReceivedApplicationMessagesCount => Interlocked.Read(ref _receivedApplicationMessagesCount); + public long SentApplicationMessagesCount => Volatile.Read(ref _statistics._sentApplicationMessagesCount); - public long ReceivedPacketsCount => Interlocked.Read(ref _receivedPacketsCount); + public long ReceivedApplicationMessagesCount => Volatile.Read(ref _statistics._receivedApplicationMessagesCount); - public long SentApplicationMessagesCount => Interlocked.Read(ref _sentApplicationMessagesCount); + public long SentPacketsCount => Volatile.Read(ref _statistics._sentPacketsCount); - public long SentPacketsCount => Interlocked.Read(ref _sentPacketsCount); + public long ReceivedPacketsCount => Volatile.Read(ref _statistics._receivedPacketsCount); public void HandleReceivedPacket(MqttPacket packet) { @@ -61,11 +60,11 @@ namespace MQTTnet.Server // This class is tracking all values from Clients perspective! LastPacketSentTimestamp = DateTime.UtcNow; - Interlocked.Increment(ref _sentPacketsCount); + Interlocked.Increment(ref _statistics._sentPacketsCount); if (packet is MqttPublishPacket) { - Interlocked.Increment(ref _sentApplicationMessagesCount); + Interlocked.Increment(ref _statistics._sentApplicationMessagesCount); } if (!(packet is MqttPingReqPacket || packet is MqttPingRespPacket)) @@ -73,6 +72,7 @@ namespace MQTTnet.Server LastNonKeepAlivePacketReceivedTimestamp = LastPacketReceivedTimestamp; } } + public void ResetStatistics() => _statistics.Reset(); public void HandleSentPacket(MqttPacket packet) { @@ -84,20 +84,29 @@ namespace MQTTnet.Server // This class is tracking all values from Clients perspective! LastPacketReceivedTimestamp = DateTime.UtcNow; - Interlocked.Increment(ref _receivedPacketsCount); + Interlocked.Increment(ref _statistics._receivedPacketsCount); if (packet is MqttPublishPacket) { - Interlocked.Increment(ref _receivedApplicationMessagesCount); + Interlocked.Increment(ref _statistics._receivedApplicationMessagesCount); } } - public void ResetStatistics() + private struct Statistics { - Interlocked.Exchange(ref _sentApplicationMessagesCount, 0); - Interlocked.Exchange(ref _receivedApplicationMessagesCount, 0); - Interlocked.Exchange(ref _sentPacketsCount, 0); - Interlocked.Exchange(ref _receivedPacketsCount, 0); + public long _receivedPacketsCount; + public long _sentPacketsCount; + + public long _receivedApplicationMessagesCount; + public long _sentApplicationMessagesCount; + + public void Reset() + { + Volatile.Write(ref _receivedPacketsCount, 0); + Volatile.Write(ref _sentPacketsCount, 0); + Volatile.Write(ref _receivedApplicationMessagesCount, 0); + Volatile.Write(ref _sentApplicationMessagesCount, 0); + } } } } \ No newline at end of file From 2aa4219838a67e336ae4f5902d956c455bc48c11 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Thu, 19 Jan 2023 20:15:15 +0100 Subject: [PATCH 2/2] Expose server session items (#1638) * Expose server session items * Add support for custom session items for injected application messages --- .github/workflows/ReleaseNotes.md | 1 + Source/MQTTnet/Adapter/MqttPacketInspector.cs | 7 +-- .../Publishing/MqttClientPublishResult.cs | 25 ++++++--- .../MqttClientPublishResultFactory.cs | 51 +++++++------------ .../Subscribing/MqttClientSubscribeResult.cs | 24 ++++++--- .../MqttClientSubscribeResultFactory.cs | 39 ++++++-------- .../MqttClientSubscribeResultItem.cs | 13 +++-- .../MqttClientUnsubscribeResult.cs | 40 ++++++++++----- .../MqttClientUnsubscribeResultFactory.cs | 42 +++++++-------- .../MqttClientUnsubscribeResultItem.cs | 22 +++++--- .../InspectMqttPacketEventArgs.cs | 10 +++- .../Events/PreparingSessionEventArgs.cs | 2 +- .../Events/ValidatingConnectionEventArgs.cs | 26 +++++----- .../Server/InjectedMqttApplicationMessage.cs | 12 ++++- .../Internal/MqttClientSessionsManager.cs | 8 ++- Source/MQTTnet/Server/MqttServer.cs | 12 +++-- 16 files changed, 185 insertions(+), 149 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index fe28bf6d..eee87661 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -5,3 +5,4 @@ * [Server] Exposed MQTT v5 sent properties from the affected client in _ClientDisconnectedAsync_ event. * [Server] Fixed wrong client ID passed to _InterceptingUnsubscriptionEventArgs_ (#1631, thanks to @ghord). * [Server] Exposed socket settings for TCP keep alive in TCP options (#1544). +* [Server] Exposed server session items at server level and allow custom session items for injected application messages (#1638). diff --git a/Source/MQTTnet/Adapter/MqttPacketInspector.cs b/Source/MQTTnet/Adapter/MqttPacketInspector.cs index c74114ae..a83de0e2 100644 --- a/Source/MQTTnet/Adapter/MqttPacketInspector.cs +++ b/Source/MQTTnet/Adapter/MqttPacketInspector.cs @@ -82,12 +82,7 @@ namespace MQTTnet.Adapter { try { - var eventArgs = new InspectMqttPacketEventArgs - { - Buffer = buffer, - Direction = direction - }; - + var eventArgs = new InspectMqttPacketEventArgs(direction, buffer); _asyncEvent.InvokeAsync(eventArgs).GetAwaiter().GetResult(); } catch (Exception exception) diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs index 6fd2dcd8..c9859d1b 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs @@ -9,9 +9,18 @@ namespace MQTTnet.Client { public sealed class MqttClientPublishResult { + public MqttClientPublishResult(ushort? packetIdentifier, MqttClientPublishReasonCode reasonCode, string reasonString, IReadOnlyCollection userProperties) + { + PacketIdentifier = packetIdentifier; + ReasonCode = reasonCode; + ReasonString = reasonString; + UserProperties = userProperties; + } + /// /// Returns if the overall status of the publish is a success. This can be the reason code _Success_ or - /// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the topic but overall transmit + /// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the + /// topic but overall transmit /// to the server etc. was a success. /// public bool IsSuccess => ReasonCode == MqttClientPublishReasonCode.Success || ReasonCode == MqttClientPublishReasonCode.NoMatchingSubscribers; @@ -19,19 +28,19 @@ namespace MQTTnet.Client /// /// Gets the packet identifier which was used for this publish. /// - public ushort? PacketIdentifier { get; set; } + public ushort? PacketIdentifier { get; } /// /// Gets or sets the reason code. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success; + public MqttClientPublishReasonCode ReasonCode { get; } = MqttClientPublishReasonCode.Success; /// /// Gets or sets the reason string. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - public string ReasonString { get; set; } + public string ReasonString { get; } /// /// Gets or sets the user properties. @@ -40,8 +49,8 @@ namespace MQTTnet.Client /// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add /// metadata to MQTT messages and pass information between publisher, broker, and subscriber. /// The feature is very similar to the HTTP header concept. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - public IReadOnlyCollection UserProperties { get; internal set; } + public IReadOnlyCollection UserProperties { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs index ea5bc233..3f4c6397 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs @@ -10,25 +10,23 @@ namespace MQTTnet.Client { public sealed class MqttClientPublishResultFactory { - static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult(); static readonly IReadOnlyCollection EmptyUserProperties = new List(); + static readonly MqttClientPublishResult AtMostOnceSuccessResult = new MqttClientPublishResult(null, MqttClientPublishReasonCode.Success, null, EmptyUserProperties); public MqttClientPublishResult Create(MqttPubAckPacket pubAckPacket) { // QoS 0 has no response. So we treat it as a success always. if (pubAckPacket == null) { - return EmptySuccessResult; + return AtMostOnceSuccessResult; } - var result = new MqttClientPublishResult - { + var result = new MqttClientPublishResult( + pubAckPacket.PacketIdentifier, // Both enums have the same values. So it can be easily converted. - ReasonCode = (MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode, - PacketIdentifier = pubAckPacket.PacketIdentifier, - ReasonString = pubAckPacket.ReasonString, - UserProperties = pubAckPacket.UserProperties ?? EmptyUserProperties - }; + (MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode, + pubAckPacket.ReasonString, + pubAckPacket.UserProperties ?? EmptyUserProperties); return result; } @@ -37,43 +35,28 @@ namespace MQTTnet.Client { if (pubRecPacket == null || pubCompPacket == null) { - return new MqttClientPublishResult - { - ReasonCode = MqttClientPublishReasonCode.UnspecifiedError - }; + var packetIdentifier = pubRecPacket?.PacketIdentifier ?? pubCompPacket?.PacketIdentifier; + return new MqttClientPublishResult(packetIdentifier, MqttClientPublishReasonCode.ImplementationSpecificError, null, EmptyUserProperties); } - MqttClientPublishResult result; - // The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC. if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound) { - result = new MqttClientPublishResult - { - PacketIdentifier = pubCompPacket.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.UnspecifiedError, - ReasonString = pubCompPacket.ReasonString, - UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties - }; - - return result; + return new MqttClientPublishResult( + pubCompPacket.PacketIdentifier, + MqttClientPublishReasonCode.UnspecifiedError, + pubCompPacket.ReasonString, + pubCompPacket.UserProperties ?? EmptyUserProperties); } - result = new MqttClientPublishResult - { - PacketIdentifier = pubCompPacket.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.Success, - ReasonString = pubCompPacket.ReasonString, - UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties - }; - + var reasonCode = MqttClientPublishReasonCode.Success; if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success) { // Both enums share the same values. - result.ReasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode; + reasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode; } - return result; + return new MqttClientPublishResult(pubCompPacket.PacketIdentifier, reasonCode, null, pubCompPacket.UserProperties ?? EmptyUserProperties); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs index 67b52600..9459679b 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; using System.Collections.Generic; using MQTTnet.Packets; @@ -9,23 +10,34 @@ namespace MQTTnet.Client { public sealed class MqttClientSubscribeResult { - public IReadOnlyCollection Items { get; internal set; } + public MqttClientSubscribeResult(ushort packetIdentifier, IReadOnlyCollection items, string reasonString, IReadOnlyCollection userProperties) + { + PacketIdentifier = packetIdentifier; + Items = items ?? throw new ArgumentNullException(nameof(items)); + ReasonString = reasonString; + UserProperties = userProperties ?? throw new ArgumentNullException(nameof(userProperties)); + } + + /// + /// Gets the result for every topic filter item. + /// + public IReadOnlyCollection Items { get; } /// /// Gets the user properties which were part of the SUBACK packet. - /// MQTTv5 only. + /// MQTT 5.0.0+ feature. /// - public IReadOnlyCollection UserProperties { get; internal set; } + public IReadOnlyCollection UserProperties { get; } /// /// Gets the reason string. - /// MQTTv5 only. + /// MQTT 5.0.0+ feature. /// - public string ReasonString { get; internal set; } + public string ReasonString { get; } /// /// Gets the packet identifier which was used. /// - public ushort PacketIdentifier { get; internal set; } + public ushort PacketIdentifier { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs index a0613248..5bb129b4 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs @@ -16,42 +16,35 @@ namespace MQTTnet.Client public MqttClientSubscribeResult Create(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) { - if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); - if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket)); - + if (subscribePacket == null) + { + throw new ArgumentNullException(nameof(subscribePacket)); + } + + if (subAckPacket == null) + { + throw new ArgumentNullException(nameof(subAckPacket)); + } + // MQTTv5.0.0 handling. if (subAckPacket.ReasonCodes.Any() && subAckPacket.ReasonCodes.Count != subscribePacket.TopicFilters.Count) { - throw new MqttProtocolViolationException( - "The reason codes are not matching the topic filters [MQTT-3.9.3-1]."); + throw new MqttProtocolViolationException("The reason codes are not matching the topic filters [MQTT-3.9.3-1]."); } - + var items = new List(); for (var i = 0; i < subscribePacket.TopicFilters.Count; i++) { items.Add(CreateSubscribeResultItem(i, subscribePacket, subAckPacket)); } - - var result = new MqttClientSubscribeResult - { - PacketIdentifier = subAckPacket.PacketIdentifier, - ReasonString = subAckPacket.ReasonString, - UserProperties = subAckPacket.UserProperties ?? EmptyUserProperties, - Items = items - }; - - return result; + + return new MqttClientSubscribeResult(subAckPacket.PacketIdentifier, items, subAckPacket.ReasonString, subAckPacket.UserProperties ?? EmptyUserProperties); } static MqttClientSubscribeResultItem CreateSubscribeResultItem(int index, MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) { - var resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[index]; - - return new MqttClientSubscribeResultItem - { - TopicFilter = subscribePacket.TopicFilters[index], - ResultCode = resultCode, - }; + var resultCode = (MqttClientSubscribeResultCode)subAckPacket.ReasonCodes[index]; + return new MqttClientSubscribeResultItem(subscribePacket.TopicFilters[index], resultCode); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs index f1fcee87..620ccffe 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs @@ -2,22 +2,29 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; using MQTTnet.Packets; namespace MQTTnet.Client { public sealed class MqttClientSubscribeResultItem { + public MqttClientSubscribeResultItem(MqttTopicFilter topicFilter, MqttClientSubscribeResultCode resultCode) + { + TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); + ResultCode = resultCode; + } + /// /// Gets or sets the topic filter. /// The topic filter can contain topics and wildcards. /// - public MqttTopicFilter TopicFilter { get; internal set; } + public MqttTopicFilter TopicFilter { get; } /// /// Gets or sets the result code. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// - public MqttClientSubscribeResultCode ResultCode { get; internal set; } + public MqttClientSubscribeResultCode ResultCode { get; } } } diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs index 6bd69401..24b8008f 100644 --- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; using System.Collections.Generic; using MQTTnet.Packets; @@ -9,23 +10,38 @@ namespace MQTTnet.Client { public sealed class MqttClientUnsubscribeResult { - public IReadOnlyCollection Items { get; internal set; } + public MqttClientUnsubscribeResult( + ushort packetIdentifier, + IReadOnlyCollection items, + string reasonString, + IReadOnlyCollection userProperties) + { + PacketIdentifier = packetIdentifier; + Items = items ?? throw new ArgumentNullException(nameof(items)); + ReasonString = reasonString; + UserProperties = userProperties ?? throw new ArgumentNullException(nameof(userProperties)); + } /// - /// Gets the user properties which were part of the UNSUBACK packet. - /// MQTTv5 only. + /// Gets the result for every topic filter item. /// - public IReadOnlyCollection UserProperties { get; internal set; } - + public IReadOnlyCollection Items { get; } + /// - /// Gets the reason string. - /// MQTTv5 only. + /// Gets the packet identifier which was used. /// - public string ReasonString { get; internal set; } - + public ushort PacketIdentifier { get; } + /// - /// Gets the packet identifier which was used. + /// Gets the reason string. + /// MQTT 5.0.0+ feature. /// - public ushort PacketIdentifier { get; internal set; } + public string ReasonString { get; } + + /// + /// Gets the user properties which were part of the UNSUBACK packet. + /// MQTT 5.0.0+ feature. + /// + public IReadOnlyCollection UserProperties { get; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs index 0f872958..d204fd94 100644 --- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs @@ -12,17 +12,23 @@ namespace MQTTnet.Client public sealed class MqttClientUnsubscribeResultFactory { static readonly IReadOnlyCollection EmptyUserProperties = new List(); - + public MqttClientUnsubscribeResult Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket) { - if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - if (unsubAckPacket == null) throw new ArgumentNullException(nameof(unsubAckPacket)); + if (unsubscribePacket == null) + { + throw new ArgumentNullException(nameof(unsubscribePacket)); + } + + if (unsubAckPacket == null) + { + throw new ArgumentNullException(nameof(unsubAckPacket)); + } // MQTTv3.1.1 has no reason code at all! if (unsubAckPacket.ReasonCodes != null && unsubAckPacket.ReasonCodes.Count != unsubscribePacket.TopicFilters.Count) { - throw new MqttProtocolViolationException( - "The return codes are not matching the topic filters [MQTT-3.9.3-1]."); + throw new MqttProtocolViolationException("The return codes are not matching the topic filters [MQTT-3.9.3-1]."); } var items = new List(); @@ -30,33 +36,21 @@ namespace MQTTnet.Client { items.Add(CreateUnsubscribeResultItem(i, unsubscribePacket, unsubAckPacket)); } - - var result = new MqttClientUnsubscribeResult - { - PacketIdentifier = unsubAckPacket.PacketIdentifier, - ReasonString = unsubAckPacket.ReasonString, - UserProperties = unsubAckPacket.UserProperties ?? EmptyUserProperties, - Items = items - }; - - return result; + + return new MqttClientUnsubscribeResult(unsubscribePacket.PacketIdentifier, items, unsubAckPacket.ReasonString, unsubAckPacket.UserProperties ?? EmptyUserProperties); } - + static MqttClientUnsubscribeResultItem CreateUnsubscribeResultItem(int index, MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket) { var resultCode = MqttClientUnsubscribeResultCode.Success; - + if (unsubAckPacket.ReasonCodes != null) { // MQTTv3.1.1 has no reason code and no return code!. - resultCode = (MqttClientUnsubscribeResultCode) unsubAckPacket.ReasonCodes[index]; + resultCode = (MqttClientUnsubscribeResultCode)unsubAckPacket.ReasonCodes[index]; } - - return new MqttClientUnsubscribeResultItem - { - TopicFilter = unsubscribePacket.TopicFilters[index], - ResultCode = resultCode - }; + + return new MqttClientUnsubscribeResultItem(unsubscribePacket.TopicFilters[index], resultCode); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs index 9f390357..997a2493 100644 --- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs @@ -2,20 +2,28 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; + namespace MQTTnet.Client { public sealed class MqttClientUnsubscribeResultItem { + public MqttClientUnsubscribeResultItem(string topicFilter, MqttClientUnsubscribeResultCode resultCode) + { + TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); + ResultCode = resultCode; + } + /// - /// Gets or sets the topic filter. - /// The topic filter can contain topics and wildcards. + /// Gets or sets the result code. + /// MQTT 5.0.0+ feature. /// - public string TopicFilter { get; internal set; } - + public MqttClientUnsubscribeResultCode ResultCode { get; } + /// - /// Gets or sets the result code. - /// Hint: MQTT 5 feature only. + /// Gets or sets the topic filter. + /// The topic filter can contain topics and wildcards. /// - public MqttClientUnsubscribeResultCode ResultCode { get; internal set; } + public string TopicFilter { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Diagnostics/PacketInspection/InspectMqttPacketEventArgs.cs b/Source/MQTTnet/Diagnostics/PacketInspection/InspectMqttPacketEventArgs.cs index 6d9e3a67..549fe863 100644 --- a/Source/MQTTnet/Diagnostics/PacketInspection/InspectMqttPacketEventArgs.cs +++ b/Source/MQTTnet/Diagnostics/PacketInspection/InspectMqttPacketEventArgs.cs @@ -8,8 +8,14 @@ namespace MQTTnet.Diagnostics { public sealed class InspectMqttPacketEventArgs : EventArgs { - public MqttPacketFlowDirection Direction { get; internal set; } + public InspectMqttPacketEventArgs(MqttPacketFlowDirection direction, byte[] buffer) + { + Direction = direction; + Buffer = buffer ?? throw new ArgumentNullException(nameof(buffer)); + } - public byte[] Buffer { get; set; } + public byte[] Buffer { get; } + + public MqttPacketFlowDirection Direction { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs b/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs index e635308a..584c4d4e 100644 --- a/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Server { public sealed class PreparingSessionEventArgs : EventArgs { - public string Id { get; internal set; } + public string Id { get; set; } // TODO: Allow adding of packets to the queue etc. diff --git a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs index 35b70369..096b2da1 100644 --- a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs @@ -9,7 +9,6 @@ using System.Security.Cryptography.X509Certificates; using System.Text; using MQTTnet.Adapter; using MQTTnet.Formatter; -using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -20,10 +19,11 @@ namespace MQTTnet.Server { readonly MqttConnectPacket _connectPacket; - public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) + public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter, IDictionary sessionItems) { _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); ChannelAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter)); + SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); } /// @@ -34,13 +34,13 @@ namespace MQTTnet.Server /// /// Gets or sets the authentication data. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// public byte[] AuthenticationData => _connectPacket.AuthenticationData; /// /// Gets or sets the authentication method. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// public string AuthenticationMethod => _connectPacket.AuthenticationMethod; @@ -99,7 +99,7 @@ namespace MQTTnet.Server /// Gets or sets the reason code. When a MQTTv3 client connects the enum value must be one which is /// also supported in MQTTv3. Otherwise the connection attempt will fail because not all codes can be /// converted properly. - /// MQTTv5 only. + /// MQTT 5.0.0+ feature. /// public MqttConnectReasonCode ReasonCode { get; set; } = MqttConnectReasonCode.Success; @@ -114,19 +114,19 @@ namespace MQTTnet.Server /// /// Gets the request problem information. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// public bool RequestProblemInformation => _connectPacket.RequestProblemInformation; /// /// Gets the request response information. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// public bool RequestResponseInformation => _connectPacket.RequestResponseInformation; /// /// Gets or sets the response authentication data. - /// MQTTv5 only. + /// MQTT 5.0.0+ feature. /// public byte[] ResponseAuthenticationData { get; set; } @@ -137,14 +137,14 @@ namespace MQTTnet.Server /// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add /// metadata to MQTT messages and pass information between publisher, broker, and subscriber. /// The feature is very similar to the HTTP header concept. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// public List ResponseUserProperties { get; set; } /// /// Gets or sets the server reference. This can be used together with i.e. "Server Moved" to send /// a different server address to the client. - /// MQTTv5 only. + /// MQTT 5.0.0+ feature. /// public string ServerReference { get; set; } @@ -158,7 +158,7 @@ namespace MQTTnet.Server /// /// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// - public IDictionary SessionItems { get; internal set; } + public IDictionary SessionItems { get; } /// /// Gets or sets the topic alias maximum. @@ -169,7 +169,7 @@ namespace MQTTnet.Server [Obsolete("This property name has a typo. Use 'UserName' instead. This one will be removed soon.")] public string Username => _connectPacket.Username; - + public string UserName => _connectPacket.Username; /// @@ -179,7 +179,7 @@ namespace MQTTnet.Server /// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add /// metadata to MQTT messages and pass information between publisher, broker, and subscriber. /// The feature is very similar to the HTTP header concept. - /// Hint: MQTT 5 feature only. + /// MQTT 5.0.0+ feature. /// public List UserProperties => _connectPacket.UserProperties; diff --git a/Source/MQTTnet/Server/InjectedMqttApplicationMessage.cs b/Source/MQTTnet/Server/InjectedMqttApplicationMessage.cs index 7d027ab6..ed829b23 100644 --- a/Source/MQTTnet/Server/InjectedMqttApplicationMessage.cs +++ b/Source/MQTTnet/Server/InjectedMqttApplicationMessage.cs @@ -3,6 +3,7 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; namespace MQTTnet.Server { @@ -13,8 +14,15 @@ namespace MQTTnet.Server ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); } - public string SenderClientId { get; set; } = string.Empty; + public MqttApplicationMessage ApplicationMessage { get; } - public MqttApplicationMessage ApplicationMessage { get; set; } + /// + /// Gets or sets the session items which should be used for all event handlers which are involved in message + /// processing. + /// If _null_ is specified the singleton session items from the server are used instead. + /// + public IDictionary CustomSessionItems { get; set; } + + public string SenderClientId { get; set; } = string.Empty; } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index b76380ab..e3dee096 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -669,11 +669,9 @@ namespace MQTTnet.Server async Task ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) { - var eventArgs = new ValidatingConnectionEventArgs(connectPacket, channelAdapter) - { - SessionItems = new ConcurrentDictionary() - }; - + // TODO: Load session items from persisted sessions in the future. + var sessionItems = new ConcurrentDictionary(); + var eventArgs = new ValidatingConnectionEventArgs(connectPacket, channelAdapter, sessionItems); await _eventContainer.ValidatingConnectionEvent.InvokeAsync(eventArgs).ConfigureAwait(false); // Check the client ID and set a random one if supported. diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 6568f0b5..d8482614 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -28,8 +28,6 @@ namespace MQTTnet.Server readonly MqttRetainedMessagesManager _retainedMessagesManager; readonly IMqttNetLogger _rootLogger; - readonly IDictionary _sessionItems = new ConcurrentDictionary(); - CancellationTokenSource _cancellationTokenSource; public MqttServer(MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) @@ -167,6 +165,12 @@ namespace MQTTnet.Server public bool IsStarted => _cancellationTokenSource != null; + /// + /// Gives access to the session items which belong to this server. This session items are passed + /// to several events instead of the client session items if the event is caused by the server instead of a client. + /// + public IDictionary ServerSessionItems { get; } = new ConcurrentDictionary(); + public Task DeleteRetainedMessagesAsync() { ThrowIfNotStarted(); @@ -228,9 +232,11 @@ namespace MQTTnet.Server throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported."); } + var sessionItems = injectedApplicationMessage.CustomSessionItems ?? ServerSessionItems; + return _clientSessionsManager.DispatchApplicationMessage( injectedApplicationMessage.SenderClientId, - _sessionItems, + sessionItems, injectedApplicationMessage.ApplicationMessage, cancellationToken); }