From 8c9460337fd31e813fda3f4b9c0a0d14509a3918 Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 10 Oct 2021 10:56:14 +0200 Subject: [PATCH] Extend interceptors. --- .../MQTTnet/Formatter/IMqttDataConverter.cs | 2 +- .../Formatter/V3/MqttV310DataConverter.cs | 31 ++--- .../Formatter/V5/MqttV500DataConverter.cs | 76 ++++++----- .../MQTTnet/Internal/CrossPlatformPromise.cs | 68 ++++++++++ Source/MQTTnet/Internal/MqttPacketBus.cs | 113 ++++++++++++++++ Source/MQTTnet/Internal/MqttPacketBusItem.cs | 45 +++++++ .../Internal/MqttPacketBusPartition.cs | 11 ++ Source/MQTTnet/Packets/MqttSubAckPacket.cs | 19 ++- Source/MQTTnet/Packets/MqttUnsubAckPacket.cs | 12 +- .../MQTTnet/Packets/MqttUnsubscribePacket.cs | 11 +- .../Server/Internal/MqttClientConnection.cs | 9 +- .../Server/Internal/MqttClientSession.cs | 7 + .../Internal/MqttClientSessionsManager.cs | 5 + .../MqttClientSubscriptionsManager.cs | 124 +++++++++++------- .../Server/Internal/UnsubscribeResult.cs | 12 ++ .../MqttSubscriptionInterceptorContext.cs | 13 +- .../MqttUnsubscriptionInterceptorContext.cs | 30 ++++- .../Server/Status/MqttSessionStatus.cs | 7 + .../{ => Internal}/AsyncLock_Tests.cs | 2 +- .../{ => Internal}/AsyncQueue_Tests.cs | 2 +- .../{ => Internal}/BlockingQueue_Tests.cs | 0 .../CrossPlatformSocket_Tests.cs | 0 .../Internal/MqttPacketBus_Tests.cs | 110 ++++++++++++++++ 23 files changed, 590 insertions(+), 119 deletions(-) create mode 100644 Source/MQTTnet/Internal/CrossPlatformPromise.cs create mode 100644 Source/MQTTnet/Internal/MqttPacketBus.cs create mode 100644 Source/MQTTnet/Internal/MqttPacketBusItem.cs create mode 100644 Source/MQTTnet/Internal/MqttPacketBusPartition.cs create mode 100644 Source/MQTTnet/Server/Internal/UnsubscribeResult.cs rename Tests/MQTTnet.Core.Tests/{ => Internal}/AsyncLock_Tests.cs (98%) rename Tests/MQTTnet.Core.Tests/{ => Internal}/AsyncQueue_Tests.cs (99%) rename Tests/MQTTnet.Core.Tests/{ => Internal}/BlockingQueue_Tests.cs (100%) rename Tests/MQTTnet.Core.Tests/{ => Internal}/CrossPlatformSocket_Tests.cs (100%) create mode 100644 Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs diff --git a/Source/MQTTnet/Formatter/IMqttDataConverter.cs b/Source/MQTTnet/Formatter/IMqttDataConverter.cs index 7f23f628..f7440b3a 100644 --- a/Source/MQTTnet/Formatter/IMqttDataConverter.cs +++ b/Source/MQTTnet/Formatter/IMqttDataConverter.cs @@ -46,7 +46,7 @@ namespace MQTTnet.Formatter MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options); - MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List reasonCodes); + MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult); MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options); } diff --git a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs index 775dc00c..89859b00 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs @@ -9,7 +9,6 @@ using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; using System; -using System.Collections.Generic; using System.Linq; using MQTTnet.Server.Internal; @@ -201,9 +200,11 @@ namespace MQTTnet.Formatter.V3 { if (options == null) throw new ArgumentNullException(nameof(options)); - var subscribePacket = new MqttSubscribePacket(); - subscribePacket.TopicFilters.AddRange(options.TopicFilters); - + var subscribePacket = new MqttSubscribePacket + { + TopicFilters = options.TopicFilters + }; + return subscribePacket; } @@ -214,11 +215,10 @@ namespace MQTTnet.Formatter.V3 var subackPacket = new MqttSubAckPacket { - PacketIdentifier = subscribePacket.PacketIdentifier + PacketIdentifier = subscribePacket.PacketIdentifier, + ReturnCodes = subscribeResult.ReturnCodes }; - - subackPacket.ReturnCodes.AddRange(subscribeResult.ReturnCodes); - + return subackPacket; } @@ -226,21 +226,22 @@ namespace MQTTnet.Formatter.V3 { if (options == null) throw new ArgumentNullException(nameof(options)); - var unsubscribePacket = new MqttUnsubscribePacket(); - unsubscribePacket.TopicFilters.AddRange(options.TopicFilters); + var unsubscribePacket = new MqttUnsubscribePacket + { + TopicFilters = options.TopicFilters + }; return unsubscribePacket; } - public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List reasonCodes) + public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult _) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes)); - + + // The UnsubscribeResult is not supported in protocol version 3.1.1! return new MqttUnsubAckPacket { - PacketIdentifier = unsubscribePacket.PacketIdentifier, - ReasonCodes = reasonCodes + PacketIdentifier = unsubscribePacket.PacketIdentifier }; } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index 5f034c0b..91b1c20b 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -49,47 +49,51 @@ namespace MQTTnet.Formatter.V5 return packet; } - public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, + MqttApplicationMessageReceivedReasonCode reasonCode) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); return new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = (MqttPubAckReasonCode)(int)reasonCode + ReasonCode = (MqttPubAckReasonCode) (int) reasonCode }; } - public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, + MqttApplicationMessageReceivedReasonCode reasonCode) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); return new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = (MqttPubRecReasonCode)(int)reasonCode + ReasonCode = (MqttPubRecReasonCode) (int) reasonCode }; } - public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, + MqttApplicationMessageReceivedReasonCode reasonCode) { if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket)); return new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier, - ReasonCode = (MqttPubCompReasonCode)(int)reasonCode + ReasonCode = (MqttPubCompReasonCode) (int) reasonCode }; } - public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, + MqttApplicationMessageReceivedReasonCode reasonCode) { if (pubRecPacket == null) throw new ArgumentNullException(nameof(pubRecPacket)); return new MqttPubRelPacket { PacketIdentifier = pubRecPacket.PacketIdentifier, - ReasonCode = (MqttPubRelReasonCode)(int)reasonCode + ReasonCode = (MqttPubRelReasonCode) (int) reasonCode }; } @@ -122,7 +126,7 @@ namespace MQTTnet.Formatter.V5 return new MqttClientConnectResult { IsSessionPresent = connAckPacket.IsSessionPresent, - ResultCode = (MqttClientConnectResultCode)(int)connAckPacket.ReasonCode, + ResultCode = (MqttClientConnectResultCode) (int) connAckPacket.ReasonCode, WildcardSubscriptionAvailable = connAckPacket.Properties?.WildcardSubscriptionAvailable ?? true, RetainAvailable = connAckPacket.Properties?.RetainAvailable ?? true, AssignedClientIdentifier = connAckPacket.Properties?.AssignedClientIdentifier, @@ -143,7 +147,8 @@ namespace MQTTnet.Formatter.V5 }; } - public MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, IMqttClientOptions options) + public MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, + IMqttClientOptions options) { if (options == null) throw new ArgumentNullException(nameof(options)); @@ -153,7 +158,7 @@ namespace MQTTnet.Formatter.V5 Username = options.Credentials?.Username, Password = options.Credentials?.Password, CleanSession = options.CleanSession, - KeepAlivePeriod = (ushort)options.KeepAlivePeriod.TotalSeconds, + KeepAlivePeriod = (ushort) options.KeepAlivePeriod.TotalSeconds, WillMessage = willApplicationMessage, Properties = new MqttConnectPacketProperties { @@ -185,7 +190,7 @@ namespace MQTTnet.Formatter.V5 SharedSubscriptionAvailable = false, TopicAliasMaximum = ushort.MaxValue, WildcardSubscriptionAvailable = true, - + UserProperties = connectionValidatorContext.ResponseUserProperties, AuthenticationMethod = connectionValidatorContext.AuthenticationMethod, AuthenticationData = connectionValidatorContext.ResponseAuthenticationData, @@ -196,38 +201,43 @@ namespace MQTTnet.Formatter.V5 }; } - public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) + public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, + MqttSubAckPacket subAckPacket) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket)); if (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 result = new MqttClientSubscribeResult(); result.Items.AddRange(subscribePacket.TopicFilters.Select((t, i) => - new MqttClientSubscribeResultItem(t, (MqttClientSubscribeResultCode)subAckPacket.ReasonCodes[i]))); + new MqttClientSubscribeResultItem(t, (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[i]))); return result; } - public MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket) + public MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, + MqttUnsubAckPacket unsubAckPacket) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); if (unsubAckPacket == null) throw new ArgumentNullException(nameof(unsubAckPacket)); if (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 result = new MqttClientUnsubscribeResult(); result.Items.AddRange(unsubscribePacket.TopicFilters.Select((t, i) => - new MqttClientUnsubscribeResultItem(t, (MqttClientUnsubscribeResultCode)unsubAckPacket.ReasonCodes[i]))); + new MqttClientUnsubscribeResultItem(t, + (MqttClientUnsubscribeResultCode) unsubAckPacket.ReasonCodes[i]))); return result; } @@ -255,11 +265,10 @@ namespace MQTTnet.Formatter.V5 var subackPacket = new MqttSubAckPacket { - PacketIdentifier = subscribePacket.PacketIdentifier + PacketIdentifier = subscribePacket.PacketIdentifier, + ReasonCodes = subscribeResult.ReasonCodes }; - subackPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes); - return subackPacket; } @@ -269,24 +278,26 @@ namespace MQTTnet.Formatter.V5 var packet = new MqttUnsubscribePacket { - Properties = new MqttUnsubscribePacketProperties() + TopicFilters = options.TopicFilters, + Properties = new MqttUnsubscribePacketProperties + { + UserProperties = options.UserProperties + } }; - packet.TopicFilters.AddRange(options.TopicFilters); - packet.Properties.UserProperties = options.UserProperties; - return packet; } - public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List reasonCodes) + public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, + UnsubscribeResult unsubscribeResult) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes)); + if (unsubscribeResult == null) throw new ArgumentNullException(nameof(unsubscribeResult)); return new MqttUnsubAckPacket { PacketIdentifier = unsubscribePacket.PacketIdentifier, - ReasonCodes = reasonCodes + ReasonCodes = unsubscribeResult.ReasonCodes }; } @@ -300,7 +311,7 @@ namespace MQTTnet.Formatter.V5 } else { - packet.ReasonCode = (MqttDisconnectReasonCode)options.ReasonCode; + packet.ReasonCode = (MqttDisconnectReasonCode) options.ReasonCode; } return packet; @@ -319,7 +330,7 @@ namespace MQTTnet.Formatter.V5 { // QoS 0 has no response. So we treat it as a success always. // Both enums have the same values. So it can be easily converted. - result.ReasonCode = (MqttClientPublishReasonCode)(int)(pubAckPacket.ReasonCode ?? 0); + result.ReasonCode = (MqttClientPublishReasonCode) (int) (pubAckPacket.ReasonCode ?? 0); result.PacketIdentifier = pubAckPacket.PacketIdentifier; } @@ -327,7 +338,8 @@ namespace MQTTnet.Formatter.V5 return result; } - public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) + public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, + MqttPubCompPacket pubCompPacket) { if (pubRecPacket == null || pubCompPacket == null) { @@ -360,7 +372,7 @@ namespace MQTTnet.Formatter.V5 if (pubRecPacket.ReasonCode.HasValue) { // Both enums share the same values. - result.ReasonCode = (MqttClientPublishReasonCode)(pubRecPacket.ReasonCode ?? 0); + result.ReasonCode = (MqttClientPublishReasonCode) (pubRecPacket.ReasonCode ?? 0); } return result; diff --git a/Source/MQTTnet/Internal/CrossPlatformPromise.cs b/Source/MQTTnet/Internal/CrossPlatformPromise.cs new file mode 100644 index 00000000..5d41b037 --- /dev/null +++ b/Source/MQTTnet/Internal/CrossPlatformPromise.cs @@ -0,0 +1,68 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Internal +{ + public sealed class CrossPlatformPromise + { + readonly TaskCompletionSource _taskCompletionSource; + + public CrossPlatformPromise() + { +#if NET452 + _taskCompletionSource = new TaskCompletionSource(); +#else + _taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); +#endif + } + + public Task Task => _taskCompletionSource.Task; + + public bool TrySetResult(TResult result) + { +#if NET452 + // To prevent deadlocks it is required to call the _TrySetResult_ method + // from a new thread because the awaiting code will not(!) be executed in + // a new thread automatically (due to await). Furthermore _this_ thread will + // do it. But _this_ thread is also reading incoming packets -> deadlock. + // NET452 does not support RunContinuationsAsynchronously + System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetResult(result)); + return true; +#else + return _taskCompletionSource.TrySetResult(result); +#endif + } + + public bool TrySetException(Exception exception) + { + if (exception == null) throw new ArgumentNullException(nameof(exception)); + +#if NET452 + // To prevent deadlocks it is required to call the _TrySetException_ method + // from a new thread because the awaiting code will not(!) be executed in + // a new thread automatically (due to await). Furthermore _this_ thread will + // do it. But _this_ thread is also reading incoming packets -> deadlock. + // NET452 does not support RunContinuationsAsynchronously + System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetException(exception)); + return true; +#else + return _taskCompletionSource.TrySetException(exception); +#endif + } + + public bool TrySetCanceled() + { +#if NET452 + // To prevent deadlocks it is required to call the _TrySetCanceled_ method + // from a new thread because the awaiting code will not(!) be executed in + // a new thread automatically (due to await). Furthermore _this_ thread will + // do it. But _this_ thread is also reading incoming packets -> deadlock. + // NET452 does not support RunContinuationsAsynchronously + System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetCanceled()); + return true; +#else + return _taskCompletionSource.TrySetCanceled(); +#endif + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Internal/MqttPacketBus.cs b/Source/MQTTnet/Internal/MqttPacketBus.cs new file mode 100644 index 00000000..43313d0d --- /dev/null +++ b/Source/MQTTnet/Internal/MqttPacketBus.cs @@ -0,0 +1,113 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Packets; + +namespace MQTTnet.Internal +{ + public sealed class MqttPacketBus : IDisposable + { + readonly object _syncRoot = new object(); + + readonly Queue[] _partitions = new Queue[] + { + new Queue(4096), + new Queue(1028), + new Queue(128) + }; + + readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0); + + int _activePartition = (int) MqttPacketBusPartition.Health; + + public int PacketsCount + { + get + { + lock (_syncRoot) + { + return _partitions.Sum(p => p.Count); + } + } + } + + public int DataPacketsCount() + { + lock (_syncRoot) + { + return _partitions[(int)MqttPacketBusPartition.Data].Count; + } + } + + public List ExportPackets(MqttPacketBusPartition partition) + { + lock (_syncRoot) + { + return _partitions[(int) partition].Select(i => i.Packet).ToList(); + } + } + + public MqttPacketBusItem Enqueue(MqttBasePacket packet, MqttPacketBusPartition partition) + { + var busItem = new MqttPacketBusItem(packet); + Enqueue(busItem, partition); + return busItem; + } + + public void Enqueue(MqttPacketBusItem item, MqttPacketBusPartition partition) + { + if (item == null) throw new ArgumentNullException(nameof(item)); + + lock (_syncRoot) + { + _partitions[(int) partition].Enqueue(item); + } + + _semaphore.Release(); + } + + void MoveActivePartition() + { + if (_activePartition >= _partitions.Length - 1) + { + _activePartition = 0; + } + else + { + _activePartition++; + } + } + + public async Task DequeueAsync(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + lock (_syncRoot) + { + for (var i = 0; i < 3; i++) + { + MoveActivePartition(); + + if (_partitions[_activePartition].Count > 0) + { + return _partitions[_activePartition].Dequeue(); + } + } + } + + // No partition contains data so that we have to wait and put + // the worker back to the thread pool. + await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + } + + throw new InvalidOperationException("MqttPacketBus is broken."); + } + + public void Dispose() + { + _semaphore?.Dispose(); + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Internal/MqttPacketBusItem.cs b/Source/MQTTnet/Internal/MqttPacketBusItem.cs new file mode 100644 index 00000000..9cfc5c5f --- /dev/null +++ b/Source/MQTTnet/Internal/MqttPacketBusItem.cs @@ -0,0 +1,45 @@ +using System; +using System.Threading.Tasks; +using MQTTnet.Packets; + +namespace MQTTnet.Internal +{ + public sealed class MqttPacketBusItem + { + readonly CrossPlatformPromise _promise = new CrossPlatformPromise(); + + public MqttPacketBusItem(MqttBasePacket packet) + { + Packet = packet ?? throw new ArgumentNullException(nameof(packet)); + } + + public MqttBasePacket Packet { get; } + + public event EventHandler Delivered; + + public Task WaitForDeliveryAsync() + { + return _promise.Task; + } + + public void MarkAsDelivered() + { + if (_promise.TrySetResult(0)) + { + Delivered?.Invoke(this, EventArgs.Empty); + } + } + + public void MarkAsFailed(Exception exception) + { + if (exception == null) throw new ArgumentNullException(nameof(exception)); + + _promise.TrySetException(exception); + } + + public void MarkAsCancelled() + { + _promise.TrySetCanceled(); + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Internal/MqttPacketBusPartition.cs b/Source/MQTTnet/Internal/MqttPacketBusPartition.cs new file mode 100644 index 00000000..298bd5bb --- /dev/null +++ b/Source/MQTTnet/Internal/MqttPacketBusPartition.cs @@ -0,0 +1,11 @@ +namespace MQTTnet.Internal +{ + public enum MqttPacketBusPartition + { + Data, + + Control, + + Health + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Packets/MqttSubAckPacket.cs b/Source/MQTTnet/Packets/MqttSubAckPacket.cs index e3848a93..5e2f6778 100644 --- a/Source/MQTTnet/Packets/MqttSubAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttSubAckPacket.cs @@ -8,15 +8,20 @@ namespace MQTTnet.Packets { public ushort PacketIdentifier { get; set; } + /// + /// Only available in MQTT v3.1.1. + /// public List ReturnCodes { get; set; } = new List(); + + /// + /// Added in MQTTv5. + /// + public List ReasonCodes { get; set; } = new List(); - #region Added in MQTTv5.0.0 - - public List ReasonCodes { get; } = new List(); - - public MqttSubAckPacketProperties Properties { get; set; } - - #endregion + /// + /// Added in MQTTv5. + /// + public MqttSubAckPacketProperties Properties { get; set; } = new MqttSubAckPacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs b/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs index 394da487..76922b94 100644 --- a/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs @@ -8,14 +8,16 @@ namespace MQTTnet.Packets { public ushort PacketIdentifier { get; set; } - #region Added in MQTTv5 - - public MqttUnsubAckPacketProperties Properties { get; set; } + /// + /// Added in MQTT V5. + /// + public MqttUnsubAckPacketProperties Properties { get; set; } = new MqttUnsubAckPacketProperties(); + /// + /// Added in MQTT V5. + /// public List ReasonCodes { get; set; } = new List(); - #endregion - public override string ToString() { var reasonCodesText = string.Join(",", ReasonCodes.Select(f => f.ToString())); diff --git a/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs b/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs index 76a9d377..8518948f 100644 --- a/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs +++ b/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs @@ -8,12 +8,11 @@ namespace MQTTnet.Packets public List TopicFilters { get; set; } = new List(); - #region Added in MQTTv5 - - public MqttUnsubscribePacketProperties Properties { get; set; } - - #endregion - + /// + /// Added in MQTTv5. + /// + public MqttUnsubscribePacketProperties Properties { get; set; } = new MqttUnsubscribePacketProperties(); + public override string ToString() { var topicFiltersText = string.Join(",", TopicFilters); diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 9143c5c3..9fe11a66 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -379,10 +379,15 @@ namespace MQTTnet.Server.Internal async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { - var reasonCodes = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket).ConfigureAwait(false); - var unsubAckPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, reasonCodes); + var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket).ConfigureAwait(false); + var unsubAckPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, unsubscribeResult); await SendPacketAsync(unsubAckPacket, cancellationToken).ConfigureAwait(false); + + if (unsubscribeResult.CloseConnection) + { + StopInternal(); + } } Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSession.cs b/Source/MQTTnet/Server/Internal/MqttClientSession.cs index 79bd558e..e03d4908 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSession.cs @@ -22,6 +22,8 @@ namespace MQTTnet.Server.Internal ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions); } + public event EventHandler Deleted; + public string ClientId { get; } public bool IsCleanSession { get; set; } = true; @@ -44,5 +46,10 @@ namespace MQTTnet.Server.Internal status.PendingApplicationMessagesCount = ApplicationMessagesQueue.Count; status.Items = Items; } + + public void OnDeleted() + { + Deleted?.Invoke(this, EventArgs.Empty); + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 5743f693..d47ad9e1 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -265,6 +265,8 @@ namespace MQTTnet.Server.Internal public async Task DeleteSessionAsync(string clientId) { MqttClientConnection connection; + MqttClientSession session; + lock (_clientConnections) { _clientConnections.TryGetValue(clientId, out connection); @@ -272,6 +274,7 @@ namespace MQTTnet.Server.Internal lock (_clientSessions) { + _clientSessions.TryGetValue(clientId, out session); _clientSessions.Remove(clientId); } @@ -280,6 +283,8 @@ namespace MQTTnet.Server.Internal await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); } + session?.OnDeleted(); + _logger.Verbose("Session for client '{0}' deleted.", clientId); } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 0dd5d2ac..3deb23da 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -29,7 +29,8 @@ namespace MQTTnet.Server.Internal _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession)); _options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher)); - _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); + _retainedMessagesManager = retainedMessagesManager ?? + throw new ArgumentNullException(nameof(retainedMessagesManager)); } public async Task Subscribe(MqttSubscribePacket subscribePacket) @@ -46,7 +47,7 @@ namespace MQTTnet.Server.Internal var interceptorContext = await InterceptSubscribe(originalTopicFilter).ConfigureAwait(false); var finalTopicFilter = interceptorContext.TopicFilter; var acceptSubscription = interceptorContext.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; - + if (string.IsNullOrEmpty(finalTopicFilter.Topic) || !acceptSubscription) { // Return codes is for MQTT < 5 and Reason Code is for MQTT > 5. @@ -69,56 +70,73 @@ namespace MQTTnet.Server.Internal continue; } - var subscription = CreateSubscription( - finalTopicFilter, - subscribePacket.Properties?.SubscriptionIdentifier ?? 0, - interceptorContext.ReasonCode); + if (interceptorContext.ProcessSubscription) + { + var subscription = CreateSubscription( + finalTopicFilter, + subscribePacket.Properties?.SubscriptionIdentifier ?? 0, + interceptorContext.ReasonCode); - await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false); + await _eventDispatcher + .SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter) + .ConfigureAwait(false); - FilterRetainedApplicationMessages(retainedApplicationMessages, subscription, result); + FilterRetainedApplicationMessages(retainedApplicationMessages, subscription, result); + } } return result; } - public async Task> Unsubscribe(MqttUnsubscribePacket unsubscribePacket) + public async Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - var reasonCodes = new List(); + var result = new UnsubscribeResult(); - foreach (var topicFilter in unsubscribePacket.TopicFilters) + _subscriptionsLock.EnterWriteLock(); + try { - var interceptorContext = await InterceptUnsubscribe(topicFilter).ConfigureAwait(false); - if (interceptorContext != null && !interceptorContext.AcceptUnsubscription) + foreach (var topicFilter in unsubscribePacket.TopicFilters) { - reasonCodes.Add(MqttUnsubscribeReasonCode.ImplementationSpecificError); - continue; - } + _subscriptions.TryGetValue(topicFilter, out var existingSubscription); + + var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription).ConfigureAwait(false); + var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success; - _subscriptionsLock.EnterWriteLock(); - try - { - reasonCodes.Add(_subscriptions.Remove(topicFilter) - ? MqttUnsubscribeReasonCode.Success - : MqttUnsubscribeReasonCode.NoSubscriptionExisted); - } - finally - { - _subscriptionsLock.ExitWriteLock(); + if (interceptorContext.CloseConnection) + { + result.CloseConnection = true; + } + + if (!acceptUnsubscription) + { + result.ReasonCodes.Add(interceptorContext.ReasonCode); + continue; + } + + if (interceptorContext.ProcessUnsubscription) + { + _subscriptions.Remove(topicFilter); + } } } + finally + { + _subscriptionsLock.ExitWriteLock(); + } foreach (var topicFilter in unsubscribePacket.TopicFilters) { - await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter) + .ConfigureAwait(false); } - return reasonCodes; + return result; } - public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel, string senderClientId) + public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel, + string senderClientId) { List subscriptions; _subscriptionsLock.EnterReadLock(); @@ -178,7 +196,8 @@ namespace MQTTnet.Server.Internal }; } - Subscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier, MqttSubscribeReasonCode reasonCode) + Subscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier, + MqttSubscribeReasonCode reasonCode) { var subscription = new Subscription { @@ -205,7 +224,7 @@ namespace MQTTnet.Server.Internal { throw new InvalidOperationException(); } - + _subscriptionsLock.EnterWriteLock(); try { @@ -220,7 +239,8 @@ namespace MQTTnet.Server.Internal return subscription; } - static void FilterRetainedApplicationMessages(IList retainedApplicationMessages, Subscription subscription, SubscribeResult subscribeResult) + static void FilterRetainedApplicationMessages(IList retainedApplicationMessages, + Subscription subscription, SubscribeResult subscribeResult) { for (var i = retainedApplicationMessages.Count - 1; i >= 0; i--) { @@ -236,13 +256,15 @@ namespace MQTTnet.Server.Internal continue; } - if (subscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly && !subscription.IsNewSubscription) + if (subscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly && + !subscription.IsNewSubscription) { // This is a MQTT V5+ feature. continue; } - if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, subscription.Topic) != MqttTopicFilterCompareResult.IsMatch) + if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, subscription.Topic) != + MqttTopicFilterCompareResult.IsMatch) { continue; } @@ -256,7 +278,7 @@ namespace MQTTnet.Server.Internal if (subscription.Identifier > 0) { - queuedApplicationMessage.SubscriptionIdentifiers = new List { subscription.Identifier }; + queuedApplicationMessage.SubscriptionIdentifiers = new List {subscription.Identifier}; } subscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage); @@ -264,7 +286,7 @@ namespace MQTTnet.Server.Internal retainedApplicationMessages[i] = null; } } - + async Task InterceptSubscribe(MqttTopicFilter topicFilter) { var context = new MqttSubscriptionInterceptorContext @@ -303,21 +325,15 @@ namespace MQTTnet.Server.Internal var interceptor = _options.SubscriptionInterceptor; if (interceptor != null) { - await interceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false); + await interceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false); } } - + return context; } - async Task InterceptUnsubscribe(string topicFilter) + async Task InterceptUnsubscribe(string topicFilter, Subscription subscription) { - var interceptor = _options.UnsubscriptionInterceptor; - if (interceptor == null) - { - return null; - } - var context = new MqttUnsubscriptionInterceptorContext { ClientId = _clientSession.ClientId, @@ -325,12 +341,28 @@ namespace MQTTnet.Server.Internal SessionItems = _clientSession.Items }; + if (subscription == null) + { + context.DefaultReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted; + } + else + { + context.DefaultReasonCode = MqttUnsubscribeReasonCode.Success; + } + + var interceptor = _options.UnsubscriptionInterceptor; + if (interceptor == null) + { + return context; + } + await interceptor.InterceptUnsubscriptionAsync(context).ConfigureAwait(false); return context; } - static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel, HashSet subscribedQoSLevels) + static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel, + HashSet subscribedQoSLevels) { MqttQualityOfServiceLevel effectiveQoS; if (subscribedQoSLevels.Contains(qosLevel)) diff --git a/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs b/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs new file mode 100644 index 00000000..37d8e002 --- /dev/null +++ b/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs @@ -0,0 +1,12 @@ +using System.Collections.Generic; +using MQTTnet.Protocol; + +namespace MQTTnet.Server.Internal +{ + public sealed class UnsubscribeResult + { + public List ReasonCodes { get; } = new List(128); + + public bool CloseConnection { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs index 28b8deb5..5b41285b 100644 --- a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs @@ -25,12 +25,13 @@ namespace MQTTnet.Server // Will be removed together with "AcceptSubscription". It only stores the default value when setting "AcceptSubscription" to true. internal MqttSubscribeReasonCode DefaultReasonCode { get; set; } + internal MqttSubscribeReturnCode DefaultReturnCode { get; set; } [Obsolete("Please use a proper value for 'ReasonCode' instead. This property will be removed in the future.")] public bool AcceptSubscription { - get => ReasonCode >= MqttSubscribeReasonCode.GrantedQoS0 && ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; + get => ReasonCode == MqttSubscribeReasonCode.GrantedQoS0 && ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; set { if (value) @@ -46,6 +47,13 @@ namespace MQTTnet.Server } } + /// + /// Gets or sets whether the broker should create an internal subscription for the client. + /// The broker can also avoid this and return "success" to the client. + /// This feature allows using the MQTT Broker as the Frontend and another system as the backend. + /// + public bool ProcessSubscription { get; set; } = true; + /// /// Gets or sets the reason code which is sent to the client. /// The subscription is skipped when the value is not GrantedQoS_. @@ -56,6 +64,9 @@ namespace MQTTnet.Server // MQTT < 5 only! internal MqttSubscribeReturnCode ReturnCode { get; set; } + /// + /// Gets or sets whether the broker should close the client connection. + /// public bool CloseConnection { get; set; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs index 99b1f1fd..9b9cb104 100644 --- a/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs @@ -1,4 +1,6 @@ -using System.Collections.Generic; +using System; +using System.Collections.Generic; +using MQTTnet.Protocol; namespace MQTTnet.Server { @@ -22,8 +24,32 @@ namespace MQTTnet.Server /// public IDictionary SessionItems { get; internal set; } - public bool AcceptUnsubscription { get; set; } = true; + [Obsolete("Please use ReasonCode instead.")] + public bool AcceptUnsubscription + { + get => ReasonCode == MqttUnsubscribeReasonCode.Success; + set => ReasonCode = DefaultReasonCode; + } + // Gets removed together with "AcceptUnsubscription". + internal MqttUnsubscribeReasonCode DefaultReasonCode { get; set; } + + /// + /// Gets or sets the reason code which will be sent to the client. + /// This value is not transmitted when using MQTT v3.1.1. + /// + public MqttUnsubscribeReasonCode ReasonCode { get; set; } + + /// + /// Gets or sets whether the broker should remove an internal subscription for the client. + /// The broker can also avoid this and return "success" to the client. + /// This feature allows using the MQTT Broker as the Frontend and another system as the backend. + /// + public bool ProcessUnsubscription { get; set; } = true; + + /// + /// Gets or sets whether the broker should close the client connection. + /// public bool CloseConnection { get; set; } } } diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index e4bd478a..a8f7b70c 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -14,8 +14,15 @@ namespace MQTTnet.Server.Status { _session = session ?? throw new ArgumentNullException(nameof(session)); _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager)); + + session.Deleted += (_, __) => Deleted?.Invoke(this, EventArgs.Empty); } + /// + /// Fired when this session is being deleted. + /// + public event EventHandler Deleted; + /// /// Gets or sets the client identifier. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. diff --git a/Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs b/Tests/MQTTnet.Core.Tests/Internal/AsyncLock_Tests.cs similarity index 98% rename from Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs rename to Tests/MQTTnet.Core.Tests/Internal/AsyncLock_Tests.cs index dfdce18f..a775e919 100644 --- a/Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Internal/AsyncLock_Tests.cs @@ -4,7 +4,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Internal; -namespace MQTTnet.Tests +namespace MQTTnet.Tests.Internal { [TestClass] public class AsyncLock_Tests diff --git a/Tests/MQTTnet.Core.Tests/AsyncQueue_Tests.cs b/Tests/MQTTnet.Core.Tests/Internal/AsyncQueue_Tests.cs similarity index 99% rename from Tests/MQTTnet.Core.Tests/AsyncQueue_Tests.cs rename to Tests/MQTTnet.Core.Tests/Internal/AsyncQueue_Tests.cs index fb1ad93a..4e0d4404 100644 --- a/Tests/MQTTnet.Core.Tests/AsyncQueue_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Internal/AsyncQueue_Tests.cs @@ -4,7 +4,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Internal; -namespace MQTTnet.Tests +namespace MQTTnet.Tests.Internal { [TestClass] public class AsyncQueue_Tests diff --git a/Tests/MQTTnet.Core.Tests/BlockingQueue_Tests.cs b/Tests/MQTTnet.Core.Tests/Internal/BlockingQueue_Tests.cs similarity index 100% rename from Tests/MQTTnet.Core.Tests/BlockingQueue_Tests.cs rename to Tests/MQTTnet.Core.Tests/Internal/BlockingQueue_Tests.cs diff --git a/Tests/MQTTnet.Core.Tests/CrossPlatformSocket_Tests.cs b/Tests/MQTTnet.Core.Tests/Internal/CrossPlatformSocket_Tests.cs similarity index 100% rename from Tests/MQTTnet.Core.Tests/CrossPlatformSocket_Tests.cs rename to Tests/MQTTnet.Core.Tests/Internal/CrossPlatformSocket_Tests.cs diff --git a/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs b/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs new file mode 100644 index 00000000..465ad1db --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs @@ -0,0 +1,110 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Internal; +using MQTTnet.Packets; + +namespace MQTTnet.Tests.Internal +{ + [TestClass] + public sealed class MqttPacketBus_Tests + { + [TestMethod] + [ExpectedException(typeof(OperationCanceledException))] + public async Task Wait_With_Empty_Bus() + { + var bus = new MqttPacketBus(); + + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1))) + { + await bus.DequeueAsync(timeout.Token); + } + } + + [TestMethod] + public void Alternate_Priorities() + { + var bus = new MqttPacketBus(); + + bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + + bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); + bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); + bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); + + bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); + bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); + bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); + + Assert.AreEqual(9, bus.PacketsCount); + + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); + Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); + + Assert.AreEqual(0, bus.PacketsCount); + } + + [TestMethod] + public void Export_Packets_Without_Dequeue() + { + var bus = new MqttPacketBus(); + + bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + + Assert.AreEqual(3, bus.PacketsCount); + + var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control); + Assert.AreEqual(0, exportedPackets.Count); + + exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health); + Assert.AreEqual(0, exportedPackets.Count); + + exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data); + Assert.AreEqual(3, exportedPackets.Count); + + Assert.AreEqual(3, bus.PacketsCount); + } + + [TestMethod] + public void Await_Single_Packet() + { + var bus = new MqttPacketBus(); + + var delivered = false; + + bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data); + bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data); + bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data).Delivered += (_, __) => + { + delivered = true; + }; + + Assert.IsFalse(delivered); + + bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered(); + + Assert.IsFalse(delivered); + + bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered(); + + Assert.IsFalse(delivered); + + bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered(); + + // The third packet has the event attached. + Assert.IsTrue(delivered); + } + } +} \ No newline at end of file