diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 781ddd0c..70594cac 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -7,7 +7,6 @@ using MQTTnet.Client.Publishing; using MQTTnet.Client.Receiving; using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; -using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Internal; using MQTTnet.PacketDispatcher; @@ -17,6 +16,7 @@ using System; using System.Threading; using System.Threading.Tasks; using MQTTnet.Diagnostics.Logger; +using MQTTnet.Formatter; using MQTTnet.Implementations; namespace MQTTnet.Client @@ -156,7 +156,8 @@ namespace MQTTnet.Client if (clientWasConnected) { - var disconnectPacket = _adapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(options); + var disconnectPacketFactory = new MqttDisconnectPacketFactory(); + var disconnectPacket = disconnectPacketFactory.Create(options); await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false); } } @@ -178,17 +179,20 @@ namespace MQTTnet.Client ThrowIfDisposed(); ThrowIfNotConnected(); - return SendAsync(new MqttAuthPacket + var authPacket = new MqttAuthPacket { - Properties = new MqttAuthPacketProperties + Properties = { // This must always be equal to the value from the CONNECT packet. So we use it here to ensure that. AuthenticationMethod = Options.AuthenticationMethod, AuthenticationData = data.AuthenticationData, - ReasonString = data.ReasonString, - UserProperties = data.UserProperties + ReasonString = data.ReasonString } - }, cancellationToken); + }; + + authPacket.Properties.UserProperties.AddRange(data.UserProperties); + + return SendAsync(authPacket, cancellationToken); } public async Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken) @@ -203,7 +207,8 @@ namespace MQTTnet.Client ThrowIfDisposed(); ThrowIfNotConnected(); - var subscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateSubscribePacket(options); + var subscribePacketFactory = new MqttSubscribePacketFactory(); + var subscribePacket = subscribePacketFactory.Create(options); subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); @@ -217,7 +222,8 @@ namespace MQTTnet.Client ThrowIfDisposed(); ThrowIfNotConnected(); - var unsubscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateUnsubscribePacket(options); + var unsubscribeFactory = new MqttUnsubscribePacketFactory(); + var unsubscribePacket = unsubscribeFactory.Create(options); unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); @@ -233,7 +239,10 @@ namespace MQTTnet.Client ThrowIfDisposed(); ThrowIfNotConnected(); - var publishPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage); + var publishPacketFactory = new MqttPublishPacketFactory(); + var publishPacket = publishPacketFactory.Create(applicationMessage); + + //_adapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage); switch (applicationMessage.QualityOfServiceLevel) { @@ -290,9 +299,8 @@ namespace MQTTnet.Client try { - var connectPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket( - willApplicationMessage, - Options); + var connectPacketFactory = new MqttConnectPacketFactory(); + var connectPacket = connectPacketFactory.Create(Options); var connAckPacket = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false); result = channelAdapter.PacketFormatterAdapter.DataConverter.CreateClientConnectResult(connAckPacket); @@ -713,7 +721,7 @@ namespace MQTTnet.Client Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) { - _disconnectReason = (MqttClientDisconnectReason) (disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection); + _disconnectReason = (MqttClientDisconnectReason) disconnectPacket.ReasonCode; // Also dispatch disconnect to waiting threads to generate a proper exception. _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket)); @@ -763,7 +771,8 @@ namespace MQTTnet.Client async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) { - var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket); + var applicationMessageFactory = new MqttApplicationMessageFactory(); + var applicationMessage = applicationMessageFactory.Create(publishPacket); var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket); var handler = ApplicationMessageReceivedHandler; diff --git a/Source/MQTTnet/Formatter/IMqttDataConverter.cs b/Source/MQTTnet/Formatter/IMqttDataConverter.cs index f7440b3a..d8f9980e 100644 --- a/Source/MQTTnet/Formatter/IMqttDataConverter.cs +++ b/Source/MQTTnet/Formatter/IMqttDataConverter.cs @@ -1,12 +1,8 @@ -using System.Collections.Generic; -using MQTTnet.Client.Connecting; -using MQTTnet.Client.Disconnecting; -using MQTTnet.Client.Options; +using MQTTnet.Client.Connecting; using MQTTnet.Client.Publishing; using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; using MQTTnet.Packets; -using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Server.Internal; @@ -14,8 +10,6 @@ namespace MQTTnet.Formatter { public interface IMqttDataConverter { - MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket); - MqttClientConnectResult CreateClientConnectResult(MqttConnAckPacket connAckPacket); MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket); @@ -25,13 +19,9 @@ namespace MQTTnet.Formatter MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket); MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket); - - MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, IMqttClientOptions options); - + MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext); - MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage); - MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode); MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode); @@ -40,14 +30,8 @@ namespace MQTTnet.Formatter MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode); - MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options); - MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult); - - MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options); - + MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult); - - MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options); } } diff --git a/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs new file mode 100644 index 00000000..c65a5dfb --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs @@ -0,0 +1,30 @@ +using System; +using MQTTnet.Packets; + +namespace MQTTnet.Formatter +{ + public sealed class MqttApplicationMessageFactory + { + public MqttApplicationMessage Create(MqttPublishPacket publishPacket) + { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + + return new MqttApplicationMessage + { + Topic = publishPacket.Topic, + Payload = publishPacket.Payload, + QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, + Retain = publishPacket.Retain, + Dup = publishPacket.Dup, + ResponseTopic = publishPacket.Properties.ResponseTopic, + ContentType = publishPacket.Properties.ContentType, + CorrelationData = publishPacket.Properties.CorrelationData, + MessageExpiryInterval = publishPacket.Properties.MessageExpiryInterval, + SubscriptionIdentifiers = publishPacket.Properties.SubscriptionIdentifiers, + TopicAlias = publishPacket.Properties.TopicAlias, + PayloadFormatIndicator = publishPacket.Properties.PayloadFormatIndicator, + UserProperties = publishPacket.Properties.UserProperties + }; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs b/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs new file mode 100644 index 00000000..defe2a96 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs @@ -0,0 +1,43 @@ +using System; +using MQTTnet.Client.Options; +using MQTTnet.Packets; + +namespace MQTTnet.Formatter +{ + public sealed class MqttConnectPacketFactory + { + public MqttConnectPacket Create(IMqttClientOptions clientOptions) + { + if (clientOptions == null) throw new ArgumentNullException(nameof(clientOptions)); + + var connectPacket = new MqttConnectPacket + { + ClientId = clientOptions.ClientId, + Username = clientOptions.Credentials?.Username, + Password = clientOptions.Credentials?.Password, + CleanSession = clientOptions.CleanSession, + KeepAlivePeriod = (ushort) clientOptions.KeepAlivePeriod.TotalSeconds, + WillMessage = clientOptions.WillMessage, + Properties = + { + AuthenticationMethod = clientOptions.AuthenticationMethod, + AuthenticationData = clientOptions.AuthenticationData, + WillDelayInterval = clientOptions.WillDelayInterval, + MaximumPacketSize = clientOptions.MaximumPacketSize, + ReceiveMaximum = clientOptions.ReceiveMaximum, + RequestProblemInformation = clientOptions.RequestProblemInformation, + RequestResponseInformation = clientOptions.RequestResponseInformation, + SessionExpiryInterval = clientOptions.SessionExpiryInterval, + TopicAliasMaximum = clientOptions.TopicAliasMaximum + } + }; + + if (clientOptions.UserProperties != null) + { + connectPacket.Properties.UserProperties.AddRange(clientOptions.UserProperties); + } + + return connectPacket; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs b/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs new file mode 100644 index 00000000..b0b3ebcf --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs @@ -0,0 +1,25 @@ +using MQTTnet.Client.Disconnecting; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter +{ + public sealed class MqttDisconnectPacketFactory + { + public MqttDisconnectPacket Create(MqttClientDisconnectOptions clientDisconnectOptions) + { + var packet = new MqttDisconnectPacket(); + + if (clientDisconnectOptions == null) + { + packet.ReasonCode = MqttDisconnectReasonCode.NormalDisconnection; + } + else + { + packet.ReasonCode = (MqttDisconnectReasonCode) clientDisconnectOptions.ReasonCode; + } + + return packet; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs new file mode 100644 index 00000000..fbcf1db3 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs @@ -0,0 +1,45 @@ +using System; +using MQTTnet.Packets; + +namespace MQTTnet.Formatter +{ + public sealed class MqttPublishPacketFactory + { + public MqttPublishPacket Create(MqttApplicationMessage applicationMessage) + { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + + // Copy all values to their matching counterparts. + // The not supported values in MQTT 3.1.1 are not serialized (excluded) later. + var packet = new MqttPublishPacket + { + Topic = applicationMessage.Topic, + Payload = applicationMessage.Payload, + QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, + Retain = applicationMessage.Retain, + Dup = applicationMessage.Dup, + Properties = + { + ContentType = applicationMessage.ContentType, + CorrelationData = applicationMessage.CorrelationData, + MessageExpiryInterval = applicationMessage.MessageExpiryInterval, + PayloadFormatIndicator = applicationMessage.PayloadFormatIndicator, + ResponseTopic = applicationMessage.ResponseTopic, + TopicAlias = applicationMessage.TopicAlias + } + }; + + if (applicationMessage.SubscriptionIdentifiers != null) + { + packet.Properties.SubscriptionIdentifiers.AddRange(applicationMessage.SubscriptionIdentifiers); + } + + if (applicationMessage.UserProperties != null) + { + packet.Properties.UserProperties.AddRange(applicationMessage.UserProperties); + } + + return packet; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttSubscribePacketFactory.cs b/Source/MQTTnet/Formatter/MqttSubscribePacketFactory.cs new file mode 100644 index 00000000..6d4710cc --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttSubscribePacketFactory.cs @@ -0,0 +1,25 @@ +using System; +using MQTTnet.Client.Subscribing; +using MQTTnet.Packets; + +namespace MQTTnet.Formatter +{ + public sealed class MqttSubscribePacketFactory + { + public MqttSubscribePacket Create(MqttClientSubscribeOptions clientSubscribeOptions) + { + if (clientSubscribeOptions == null) throw new ArgumentNullException(nameof(clientSubscribeOptions)); + + var packet = new MqttSubscribePacket(); + packet.TopicFilters.AddRange(clientSubscribeOptions.TopicFilters); + packet.Properties.SubscriptionIdentifier = clientSubscribeOptions.SubscriptionIdentifier; + + if (clientSubscribeOptions.UserProperties != null) + { + packet.Properties.UserProperties.AddRange(clientSubscribeOptions.UserProperties); + } + + return packet; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttUnsubscribePacketFactory.cs b/Source/MQTTnet/Formatter/MqttUnsubscribePacketFactory.cs new file mode 100644 index 00000000..ade26897 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttUnsubscribePacketFactory.cs @@ -0,0 +1,28 @@ +using System; +using MQTTnet.Client.Unsubscribing; +using MQTTnet.Packets; + +namespace MQTTnet.Formatter +{ + public sealed class MqttUnsubscribePacketFactory + { + public MqttUnsubscribePacket Create(MqttClientUnsubscribeOptions clientUnsubscribeOptions) + { + if (clientUnsubscribeOptions == null) throw new ArgumentNullException(nameof(clientUnsubscribeOptions)); + + var packet = new MqttUnsubscribePacket(); + + if (clientUnsubscribeOptions.TopicFilters != null) + { + packet.TopicFilters.AddRange(clientUnsubscribeOptions.TopicFilters); + } + + if (clientUnsubscribeOptions.UserProperties != null) + { + packet.Properties.UserProperties.AddRange(clientUnsubscribeOptions.UserProperties); + } + + return packet; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs index 89859b00..db7b7406 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs @@ -1,6 +1,4 @@ using MQTTnet.Client.Connecting; -using MQTTnet.Client.Disconnecting; -using MQTTnet.Client.Options; using MQTTnet.Client.Publishing; using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; @@ -16,20 +14,6 @@ namespace MQTTnet.Formatter.V3 { public class MqttV310DataConverter : IMqttDataConverter { - public MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage) - { - if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - - return new MqttPublishPacket - { - Topic = applicationMessage.Topic, - Payload = applicationMessage.Payload, - QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, - Retain = applicationMessage.Retain, - Dup = false - }; - } - public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); @@ -69,21 +53,7 @@ namespace MQTTnet.Formatter.V3 PacketIdentifier = pubRecPacket.PacketIdentifier }; } - - public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - return new MqttApplicationMessage - { - Topic = publishPacket.Topic, - Payload = publishPacket.Payload, - QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, - Retain = publishPacket.Retain, - Dup = publishPacket.Dup - }; - } - + public MqttClientConnectResult CreateClientConnectResult(MqttConnAckPacket connAckPacket) { if (connAckPacket == null) throw new ArgumentNullException(nameof(connAckPacket)); @@ -139,22 +109,7 @@ namespace MQTTnet.Formatter.V3 ResultCode = resultCode }; } - - public MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, IMqttClientOptions options) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - return new MqttConnectPacket - { - ClientId = options.ClientId, - Username = options.Credentials?.Username, - Password = options.Credentials?.Password, - CleanSession = options.CleanSession, - KeepAlivePeriod = (ushort)options.KeepAlivePeriod.TotalSeconds, - WillMessage = willApplicationMessage - }; - } - + public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext) { if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext)); @@ -195,45 +150,22 @@ namespace MQTTnet.Formatter.V3 return result; } - - public MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - var subscribePacket = new MqttSubscribePacket - { - TopicFilters = options.TopicFilters - }; - - return subscribePacket; - } - + public MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult)); - var subackPacket = new MqttSubAckPacket + var subAckPacket = new MqttSubAckPacket { - PacketIdentifier = subscribePacket.PacketIdentifier, - ReturnCodes = subscribeResult.ReturnCodes + PacketIdentifier = subscribePacket.PacketIdentifier }; - return subackPacket; + subAckPacket.ReturnCodes.AddRange(subscribeResult.ReturnCodes); + + return subAckPacket; } - - public MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - var unsubscribePacket = new MqttUnsubscribePacket - { - TopicFilters = options.TopicFilters - }; - - return unsubscribePacket; - } - + public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult _) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); @@ -244,12 +176,7 @@ namespace MQTTnet.Formatter.V3 PacketIdentifier = unsubscribePacket.PacketIdentifier }; } - - public MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options) - { - return new MqttDisconnectPacket(); - } - + public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket) { return new MqttClientPublishResult diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index 91b1c20b..410cc869 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -1,6 +1,4 @@ using MQTTnet.Client.Connecting; -using MQTTnet.Client.Disconnecting; -using MQTTnet.Client.Options; using MQTTnet.Client.Publishing; using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; @@ -17,38 +15,6 @@ namespace MQTTnet.Formatter.V5 { public sealed class MqttV500DataConverter : IMqttDataConverter { - public MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage) - { - if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - - var packet = new MqttPublishPacket - { - Topic = applicationMessage.Topic, - Payload = applicationMessage.Payload, - QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, - Retain = applicationMessage.Retain, - Dup = applicationMessage.Dup, - Properties = new MqttPublishPacketProperties - { - ContentType = applicationMessage.ContentType, - CorrelationData = applicationMessage.CorrelationData, - MessageExpiryInterval = applicationMessage.MessageExpiryInterval, - PayloadFormatIndicator = applicationMessage.PayloadFormatIndicator, - ResponseTopic = applicationMessage.ResponseTopic, - SubscriptionIdentifiers = applicationMessage.SubscriptionIdentifiers, - TopicAlias = applicationMessage.TopicAlias - } - }; - - if (applicationMessage.UserProperties != null) - { - packet.Properties.UserProperties = new List(); - packet.Properties.UserProperties.AddRange(applicationMessage.UserProperties); - } - - return packet; - } - public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { @@ -97,28 +63,6 @@ namespace MQTTnet.Formatter.V5 }; } - public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - return new MqttApplicationMessage - { - Topic = publishPacket.Topic, - Payload = publishPacket.Payload, - QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, - Retain = publishPacket.Retain, - Dup = publishPacket.Dup, - ResponseTopic = publishPacket.Properties?.ResponseTopic, - ContentType = publishPacket.Properties?.ContentType, - CorrelationData = publishPacket.Properties?.CorrelationData, - MessageExpiryInterval = publishPacket.Properties?.MessageExpiryInterval, - SubscriptionIdentifiers = publishPacket.Properties?.SubscriptionIdentifiers, - TopicAlias = publishPacket.Properties?.TopicAlias, - PayloadFormatIndicator = publishPacket.Properties?.PayloadFormatIndicator, - UserProperties = publishPacket.Properties?.UserProperties ?? new List() - }; - } - public MqttClientConnectResult CreateClientConnectResult(MqttConnAckPacket connAckPacket) { if (connAckPacket == null) throw new ArgumentNullException(nameof(connAckPacket)); @@ -146,52 +90,22 @@ namespace MQTTnet.Formatter.V5 UserProperties = connAckPacket.Properties?.UserProperties ?? new List() }; } - - public MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, - IMqttClientOptions options) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - return new MqttConnectPacket - { - ClientId = options.ClientId, - Username = options.Credentials?.Username, - Password = options.Credentials?.Password, - CleanSession = options.CleanSession, - KeepAlivePeriod = (ushort) options.KeepAlivePeriod.TotalSeconds, - WillMessage = willApplicationMessage, - Properties = new MqttConnectPacketProperties - { - AuthenticationMethod = options.AuthenticationMethod, - AuthenticationData = options.AuthenticationData, - WillDelayInterval = options.WillDelayInterval, - MaximumPacketSize = options.MaximumPacketSize, - ReceiveMaximum = options.ReceiveMaximum, - RequestProblemInformation = options.RequestProblemInformation, - RequestResponseInformation = options.RequestResponseInformation, - SessionExpiryInterval = options.SessionExpiryInterval, - TopicAliasMaximum = options.TopicAliasMaximum, - UserProperties = options.UserProperties - } - }; - } - + public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext) { if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext)); - return new MqttConnAckPacket + var connAckPacket = new MqttConnAckPacket { ReasonCode = connectionValidatorContext.ReasonCode, - Properties = new MqttConnAckPacketProperties + Properties = { RetainAvailable = true, SubscriptionIdentifiersAvailable = true, SharedSubscriptionAvailable = false, TopicAliasMaximum = ushort.MaxValue, WildcardSubscriptionAvailable = true, - - UserProperties = connectionValidatorContext.ResponseUserProperties, + AuthenticationMethod = connectionValidatorContext.AuthenticationMethod, AuthenticationData = connectionValidatorContext.ResponseAuthenticationData, AssignedClientIdentifier = connectionValidatorContext.AssignedClientIdentifier, @@ -199,6 +113,13 @@ namespace MQTTnet.Formatter.V5 ServerReference = connectionValidatorContext.ServerReference } }; + + if (connectionValidatorContext.ResponseUserProperties != null) + { + connAckPacket.Properties.UserProperties.AddRange(connectionValidatorContext.ResponseUserProperties); + } + + return connAckPacket; } public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, @@ -241,82 +162,38 @@ namespace MQTTnet.Formatter.V5 return result; } - - public MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - var packet = new MqttSubscribePacket - { - Properties = new MqttSubscribePacketProperties() - }; - - packet.TopicFilters.AddRange(options.TopicFilters); - packet.Properties.SubscriptionIdentifier = options.SubscriptionIdentifier; - packet.Properties.UserProperties = options.UserProperties; - - return packet; - } - + public MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult)); - var subackPacket = new MqttSubAckPacket + var subAckPacket = new MqttSubAckPacket { - PacketIdentifier = subscribePacket.PacketIdentifier, - ReasonCodes = subscribeResult.ReasonCodes + PacketIdentifier = subscribePacket.PacketIdentifier }; - return subackPacket; + subAckPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes); + + return subAckPacket; } - - public MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - var packet = new MqttUnsubscribePacket - { - TopicFilters = options.TopicFilters, - Properties = new MqttUnsubscribePacketProperties - { - UserProperties = options.UserProperties - } - }; - - return packet; - } - + public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); if (unsubscribeResult == null) throw new ArgumentNullException(nameof(unsubscribeResult)); - return new MqttUnsubAckPacket + var unsubAckPacket = new MqttUnsubAckPacket { - PacketIdentifier = unsubscribePacket.PacketIdentifier, - ReasonCodes = unsubscribeResult.ReasonCodes + PacketIdentifier = unsubscribePacket.PacketIdentifier }; + + unsubAckPacket.ReasonCodes.AddRange(unsubscribeResult.ReasonCodes); + + return unsubAckPacket; } - - public MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options) - { - var packet = new MqttDisconnectPacket(); - - if (options == null) - { - packet.ReasonCode = MqttDisconnectReasonCode.NormalDisconnection; - } - else - { - packet.ReasonCode = (MqttDisconnectReasonCode) options.ReasonCode; - } - - return packet; - } - + public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket) { var result = new MqttClientPublishResult @@ -330,7 +207,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; result.PacketIdentifier = pubAckPacket.PacketIdentifier; } @@ -369,10 +246,10 @@ namespace MQTTnet.Formatter.V5 UserProperties = pubCompPacket.Properties?.UserProperties }; - if (pubRecPacket.ReasonCode.HasValue) + if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success) { // Both enums share the same values. - result.ReasonCode = (MqttClientPublishReasonCode) (pubRecPacket.ReasonCode ?? 0); + result.ReasonCode = (MqttClientPublishReasonCode) pubRecPacket.ReasonCode; } return result; diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs index 6fc6365e..fb240645 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs @@ -84,11 +84,6 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttConnectPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.SessionExpiryInterval) { packet.Properties.SessionExpiryInterval = propertiesReader.ReadSessionExpiryInterval(); @@ -123,11 +118,6 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -223,15 +213,17 @@ namespace MQTTnet.Formatter.V5 var packet = new MqttConnAckPacket { IsSessionPresent = (acknowledgeFlags & 0x1) > 0, - ReasonCode = (MqttConnectReasonCode)body.ReadByte() + ReasonCode = (MqttConnectReasonCode)body.ReadByte(), + Properties = + { + // indicate that a feature is available. + // Set all default values according to specification. When they are missing the often + RetainAvailable = true, + SharedSubscriptionAvailable = true, + SubscriptionIdentifiersAvailable = true, + WildcardSubscriptionAvailable = true + } }; - - // Set all default values according to specification. When they are missing the often - // indicate that a feature is available. - packet.Properties.RetainAvailable = true; - packet.Properties.SharedSubscriptionAvailable = true; - packet.Properties.SubscriptionIdentifiersAvailable = true; - packet.Properties.WildcardSubscriptionAvailable = true; var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) @@ -302,11 +294,6 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -330,11 +317,6 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttDisconnectPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.SessionExpiryInterval) { packet.Properties.SessionExpiryInterval = propertiesReader.ReadSessionExpiryInterval(); @@ -349,11 +331,6 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -377,22 +354,12 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttSubscribePacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier) { packet.Properties.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier(); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -436,22 +403,12 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttSubAckPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.ReasonString) { packet.Properties.ReasonString = propertiesReader.ReadReasonString(); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -481,18 +438,8 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttUnsubscribePacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -521,22 +468,12 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttUnsubAckPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.ReasonString) { packet.Properties.ReasonString = propertiesReader.ReadReasonString(); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -588,11 +525,6 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttPublishPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.PayloadFormatIndicator) { packet.Properties.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator(); @@ -615,11 +547,6 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier) { - if (packet.Properties.SubscriptionIdentifiers == null) - { - packet.Properties.SubscriptionIdentifiers = new List(); - } - packet.Properties.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier()); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.ContentType) @@ -628,11 +555,6 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -669,22 +591,12 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttPubAckPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.ReasonString) { packet.Properties.ReasonString = propertiesReader.ReadReasonString(); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -716,22 +628,12 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttPubRecPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.ReasonString) { packet.Properties.ReasonString = propertiesReader.ReadReasonString(); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -763,22 +665,12 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttPubRelPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.ReasonString) { packet.Properties.ReasonString = propertiesReader.ReadReasonString(); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -810,22 +702,12 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttPubCompPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.ReasonString) { packet.Properties.ReasonString = propertiesReader.ReadReasonString(); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else @@ -854,11 +736,6 @@ namespace MQTTnet.Formatter.V5 var propertiesReader = new MqttV500PropertiesReader(body); while (propertiesReader.MoveNext()) { - if (packet.Properties == null) - { - packet.Properties = new MqttAuthPacketProperties(); - } - if (propertiesReader.CurrentPropertyId == MqttPropertyId.AuthenticationMethod) { packet.Properties.AuthenticationMethod = propertiesReader.ReadAuthenticationMethod(); @@ -873,11 +750,6 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.Properties.UserProperties == null) - { - packet.Properties.UserProperties = new List(); - } - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index abfc5271..141346a8 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -289,12 +289,7 @@ namespace MQTTnet.Formatter.V5 { throw new MqttProtocolViolationException("PubAck packet has no packet identifier."); } - - if (!packet.ReasonCode.HasValue) - { - throw new MqttProtocolViolationException("PubAck packet must contain a reason code."); - } - + packetWriter.Write(packet.PacketIdentifier); var propertiesWriter = new MqttV500PropertiesWriter(); @@ -304,9 +299,9 @@ namespace MQTTnet.Formatter.V5 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties); } - if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubAckReasonCode.Success) + if (packetWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success) { - packetWriter.Write((byte)packet.ReasonCode.Value); + packetWriter.Write((byte)packet.ReasonCode); propertiesWriter.WriteTo(packetWriter); } @@ -316,12 +311,7 @@ namespace MQTTnet.Formatter.V5 static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter) { ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - if (!packet.ReasonCode.HasValue) - { - ThrowReasonCodeNotSetException(); - } - + var propertiesWriter = new MqttV500PropertiesWriter(); if (packet.Properties != null) { @@ -331,9 +321,9 @@ namespace MQTTnet.Formatter.V5 packetWriter.Write(packet.PacketIdentifier); - if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRecReasonCode.Success) + if (packetWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success) { - packetWriter.Write((byte)packet.ReasonCode.Value); + packetWriter.Write((byte)packet.ReasonCode); propertiesWriter.WriteTo(packetWriter); } @@ -343,12 +333,7 @@ namespace MQTTnet.Formatter.V5 static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter) { ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - if (!packet.ReasonCode.HasValue) - { - ThrowReasonCodeNotSetException(); - } - + var propertiesWriter = new MqttV500PropertiesWriter(); if (packet.Properties != null) { @@ -358,9 +343,9 @@ namespace MQTTnet.Formatter.V5 packetWriter.Write(packet.PacketIdentifier); - if (propertiesWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRelReasonCode.Success) + if (propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRelReasonCode.Success) { - packetWriter.Write((byte)packet.ReasonCode.Value); + packetWriter.Write((byte)packet.ReasonCode); propertiesWriter.WriteTo(packetWriter); } @@ -370,12 +355,7 @@ namespace MQTTnet.Formatter.V5 static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter) { ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - if (!packet.ReasonCode.HasValue) - { - ThrowReasonCodeNotSetException(); - } - + packetWriter.Write(packet.PacketIdentifier); var propertiesWriter = new MqttV500PropertiesWriter(); @@ -385,9 +365,9 @@ namespace MQTTnet.Formatter.V5 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties); } - if (propertiesWriter.Length > 0 || packet.ReasonCode.Value != MqttPubCompReasonCode.Success) + if (propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubCompReasonCode.Success) { - packetWriter.Write((byte)packet.ReasonCode.Value); + packetWriter.Write((byte)packet.ReasonCode); propertiesWriter.WriteTo(packetWriter); } @@ -521,12 +501,7 @@ namespace MQTTnet.Formatter.V5 static byte EncodeDisconnectPacket(MqttDisconnectPacket packet, IMqttPacketWriter packetWriter) { - if (!packet.ReasonCode.HasValue) - { - ThrowReasonCodeNotSetException(); - } - - packetWriter.Write((byte)packet.ReasonCode.Value); + packetWriter.Write((byte)packet.ReasonCode); var propertiesWriter = new MqttV500PropertiesWriter(); if (packet.Properties != null) @@ -569,12 +544,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Auth); } - - static void ThrowReasonCodeNotSetException() - { - throw new MqttProtocolViolationException("The ReasonCode must be set for MQTT version 5."); - } - + static void ThrowIfPacketIdentifierIsInvalid(ushort packetIdentifier, MqttBasePacket packet) { // SUBSCRIBE, UNSUBSCRIBE, and PUBLISH(in cases where QoS > 0) Control Packets MUST contain a non-zero 16 - bit Packet Identifier[MQTT - 2.3.1 - 1]. diff --git a/Source/MQTTnet/Packets/MqttAuthPacket.cs b/Source/MQTTnet/Packets/MqttAuthPacket.cs index a7925709..5357c633 100644 --- a/Source/MQTTnet/Packets/MqttAuthPacket.cs +++ b/Source/MQTTnet/Packets/MqttAuthPacket.cs @@ -9,6 +9,6 @@ namespace MQTTnet.Packets { public MqttAuthenticateReasonCode ReasonCode { get; set; } - public MqttAuthPacketProperties Properties { get; set; } + public MqttAuthPacketProperties Properties { get; } = new MqttAuthPacketProperties(); } } diff --git a/Source/MQTTnet/Packets/MqttAuthPacketProperties.cs b/Source/MQTTnet/Packets/MqttAuthPacketProperties.cs index d4910af1..ce4a989f 100644 --- a/Source/MQTTnet/Packets/MqttAuthPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttAuthPacketProperties.cs @@ -10,6 +10,6 @@ namespace MQTTnet.Packets public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttConnAckPacket.cs b/Source/MQTTnet/Packets/MqttConnAckPacket.cs index b8db02d2..4a054e3d 100644 --- a/Source/MQTTnet/Packets/MqttConnAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnAckPacket.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Packets /// /// Added in MQTT 5.0.0. /// - public MqttConnAckPacketProperties Properties { get; set; } = new MqttConnAckPacketProperties(); + public MqttConnAckPacketProperties Properties { get; } = new MqttConnAckPacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs index 519512b9..a0e0d7ae 100644 --- a/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs @@ -21,7 +21,7 @@ namespace MQTTnet.Packets public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); public bool WildcardSubscriptionAvailable { get; set; } diff --git a/Source/MQTTnet/Packets/MqttConnectPacket.cs b/Source/MQTTnet/Packets/MqttConnectPacket.cs index 61302628..dcc6d447 100644 --- a/Source/MQTTnet/Packets/MqttConnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnectPacket.cs @@ -10,17 +10,18 @@ public ushort KeepAlivePeriod { get; set; } - // Also called "Clean Start" in MQTTv5. + /// + /// Also called "Clean Start" in MQTTv5. + /// public bool CleanSession { get; set; } public MqttApplicationMessage WillMessage { get; set; } - - #region Added in MQTTv5.0.0 - - public MqttConnectPacketProperties Properties { get; set; } - - #endregion - + + /// + /// Added in MQTTv5. + /// + public MqttConnectPacketProperties Properties { get; } = new MqttConnectPacketProperties(); + public override string ToString() { var passwordText = string.Empty; diff --git a/Source/MQTTnet/Packets/MqttConnectPacketProperties.cs b/Source/MQTTnet/Packets/MqttConnectPacketProperties.cs index b9bd81ea..cbf0cba3 100644 --- a/Source/MQTTnet/Packets/MqttConnectPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttConnectPacketProperties.cs @@ -22,6 +22,6 @@ namespace MQTTnet.Packets public uint? MaximumPacketSize { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Packets/MqttDisconnectPacket.cs b/Source/MQTTnet/Packets/MqttDisconnectPacket.cs index d28e57b8..64e9451c 100644 --- a/Source/MQTTnet/Packets/MqttDisconnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttDisconnectPacket.cs @@ -4,13 +4,15 @@ namespace MQTTnet.Packets { public sealed class MqttDisconnectPacket : MqttBasePacket { - #region Added in MQTTv5 + /// + /// Added in MQTTv5. + /// + public MqttDisconnectReasonCode ReasonCode { get; set; } = MqttDisconnectReasonCode.NormalDisconnection; - public MqttDisconnectReasonCode? ReasonCode { get; set; } - - public MqttDisconnectPacketProperties Properties { get; set; } - - #endregion + /// + /// Added in MQTTv5. + /// + public MqttDisconnectPacketProperties Properties { get; } = new MqttDisconnectPacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttDisconnectPacketProperties.cs b/Source/MQTTnet/Packets/MqttDisconnectPacketProperties.cs index c4f8f794..ca826f2b 100644 --- a/Source/MQTTnet/Packets/MqttDisconnectPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttDisconnectPacketProperties.cs @@ -8,7 +8,7 @@ namespace MQTTnet.Packets public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); public string ServerReference { get; set; } } diff --git a/Source/MQTTnet/Packets/MqttPubAckPacket.cs b/Source/MQTTnet/Packets/MqttPubAckPacket.cs index dbd8d713..1a87129e 100644 --- a/Source/MQTTnet/Packets/MqttPubAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttPubAckPacket.cs @@ -6,13 +6,15 @@ namespace MQTTnet.Packets { public ushort PacketIdentifier { get; set; } - #region Added in MQTTv5 + /// + /// Added in MQTTv5. + /// + public MqttPubAckReasonCode ReasonCode { get; set; } = MqttPubAckReasonCode.Success; - public MqttPubAckReasonCode? ReasonCode { get; set; } - - public MqttPubAckPacketProperties Properties { get; set; } - - #endregion + /// + /// Added in MQTTv5. + /// + public MqttPubAckPacketProperties Properties { get; } = new MqttPubAckPacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttPubAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttPubAckPacketProperties.cs index 58cb0572..737b4b01 100644 --- a/Source/MQTTnet/Packets/MqttPubAckPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttPubAckPacketProperties.cs @@ -6,6 +6,6 @@ namespace MQTTnet.Packets { public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttPubCompPacket.cs b/Source/MQTTnet/Packets/MqttPubCompPacket.cs index 573b0eb6..74fec09a 100644 --- a/Source/MQTTnet/Packets/MqttPubCompPacket.cs +++ b/Source/MQTTnet/Packets/MqttPubCompPacket.cs @@ -5,15 +5,17 @@ namespace MQTTnet.Packets public sealed class MqttPubCompPacket : MqttBasePacket, IMqttPacketWithIdentifier { public ushort PacketIdentifier { get; set; } + + /// + /// Added in MQTTv5. + /// + public MqttPubCompReasonCode ReasonCode { get; set; } = MqttPubCompReasonCode.Success; - #region Added in MQTTv5 - - public MqttPubCompReasonCode? ReasonCode { get; set; } - - public MqttPubCompPacketProperties Properties { get; set; } - - #endregion - + /// + /// Added in MQTTv5. + /// + public MqttPubCompPacketProperties Properties { get; } = new MqttPubCompPacketProperties(); + public override string ToString() { return string.Concat("PubComp: [PacketIdentifier=", PacketIdentifier, "] [ReasonCode=", ReasonCode, "]"); diff --git a/Source/MQTTnet/Packets/MqttPubCompPacketProperties.cs b/Source/MQTTnet/Packets/MqttPubCompPacketProperties.cs index 9d2e1a75..cfbe0c96 100644 --- a/Source/MQTTnet/Packets/MqttPubCompPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttPubCompPacketProperties.cs @@ -6,6 +6,6 @@ namespace MQTTnet.Packets { public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttPubRecPacket.cs b/Source/MQTTnet/Packets/MqttPubRecPacket.cs index 7e704bec..b05fc7b3 100644 --- a/Source/MQTTnet/Packets/MqttPubRecPacket.cs +++ b/Source/MQTTnet/Packets/MqttPubRecPacket.cs @@ -6,13 +6,15 @@ namespace MQTTnet.Packets { public ushort PacketIdentifier { get; set; } - #region Added in MQTTv5 + /// + /// Added in MQTTv5. + /// + public MqttPubRecReasonCode ReasonCode { get; set; } = MqttPubRecReasonCode.Success; - public MqttPubRecReasonCode? ReasonCode { get; set; } - - public MqttPubRecPacketProperties Properties { get; set; } - - #endregion + /// + /// Added in MQTTv5. + /// + public MqttPubRecPacketProperties Properties { get; } = new MqttPubRecPacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttPubRecPacketProperties.cs b/Source/MQTTnet/Packets/MqttPubRecPacketProperties.cs index 2d302e82..c40067f1 100644 --- a/Source/MQTTnet/Packets/MqttPubRecPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttPubRecPacketProperties.cs @@ -6,6 +6,6 @@ namespace MQTTnet.Packets { public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttPubRelPacket.cs b/Source/MQTTnet/Packets/MqttPubRelPacket.cs index 810c3ba7..05c40121 100644 --- a/Source/MQTTnet/Packets/MqttPubRelPacket.cs +++ b/Source/MQTTnet/Packets/MqttPubRelPacket.cs @@ -6,13 +6,15 @@ namespace MQTTnet.Packets { public ushort PacketIdentifier { get; set; } - #region Added in MQTTv5 + /// + /// Added in MQTTv5. + /// + public MqttPubRelReasonCode ReasonCode { get; set; } = MqttPubRelReasonCode.Success; - public MqttPubRelReasonCode? ReasonCode { get; set; } - - public MqttPubRelPacketProperties Properties { get; set; } - - #endregion + /// + /// Added in MQTTv5. + /// + public MqttPubRelPacketProperties Properties { get; } = new MqttPubRelPacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttPubRelPacketProperties.cs b/Source/MQTTnet/Packets/MqttPubRelPacketProperties.cs index 9cd610b3..6651da89 100644 --- a/Source/MQTTnet/Packets/MqttPubRelPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttPubRelPacketProperties.cs @@ -6,6 +6,6 @@ namespace MQTTnet.Packets { public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttPublishPacket.cs b/Source/MQTTnet/Packets/MqttPublishPacket.cs index 9fabdf9f..4fad06d3 100644 --- a/Source/MQTTnet/Packets/MqttPublishPacket.cs +++ b/Source/MQTTnet/Packets/MqttPublishPacket.cs @@ -16,12 +16,11 @@ namespace MQTTnet.Packets public byte[] Payload { get; set; } - #region Added in MQTTv5 - - public MqttPublishPacketProperties Properties { get; set; } - - #endregion - + /// + /// Added in MQTTv5. + /// + public MqttPublishPacketProperties Properties { get; } = new MqttPublishPacketProperties(); + public override string ToString() { return string.Concat("Publish: [Topic=", Topic, "] [Payload.Length=", Payload?.Length, "] [QoSLevel=", QualityOfServiceLevel, "] [Dup=", Dup, "] [Retain=", Retain, "] [PacketIdentifier=", PacketIdentifier, "]"); diff --git a/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs index 076edf57..0278d909 100644 --- a/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs @@ -15,9 +15,9 @@ namespace MQTTnet.Packets public byte[] CorrelationData { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); - public List SubscriptionIdentifiers { get; set; } + public List SubscriptionIdentifiers { get; } = new List(); public string ContentType { get; set; } } diff --git a/Source/MQTTnet/Packets/MqttSubAckPacket.cs b/Source/MQTTnet/Packets/MqttSubAckPacket.cs index 5e2f6778..c7c2242d 100644 --- a/Source/MQTTnet/Packets/MqttSubAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttSubAckPacket.cs @@ -11,17 +11,17 @@ namespace MQTTnet.Packets /// /// Only available in MQTT v3.1.1. /// - public List ReturnCodes { get; set; } = new List(); + public List ReturnCodes { get; } = new List(); /// /// Added in MQTTv5. /// - public List ReasonCodes { get; set; } = new List(); + public List ReasonCodes { get; } = new List(); /// /// Added in MQTTv5. /// - public MqttSubAckPacketProperties Properties { get; set; } = new MqttSubAckPacketProperties(); + public MqttSubAckPacketProperties Properties { get; } = new MqttSubAckPacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttSubAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttSubAckPacketProperties.cs index a8a2f571..d33091ae 100644 --- a/Source/MQTTnet/Packets/MqttSubAckPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttSubAckPacketProperties.cs @@ -6,6 +6,6 @@ namespace MQTTnet.Packets { public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttSubscribePacket.cs b/Source/MQTTnet/Packets/MqttSubscribePacket.cs index 3743c9c2..ada833d0 100644 --- a/Source/MQTTnet/Packets/MqttSubscribePacket.cs +++ b/Source/MQTTnet/Packets/MqttSubscribePacket.cs @@ -7,12 +7,12 @@ namespace MQTTnet.Packets { public ushort PacketIdentifier { get; set; } - public List TopicFilters { get; set; } = new List(); + public List TopicFilters { get; } = new List(); /// /// Added in MQTT V5. /// - public MqttSubscribePacketProperties Properties { get; set; } = new MqttSubscribePacketProperties(); + public MqttSubscribePacketProperties Properties { get; } = new MqttSubscribePacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttSubscribePacketProperties.cs b/Source/MQTTnet/Packets/MqttSubscribePacketProperties.cs index 8455f890..86e0275f 100644 --- a/Source/MQTTnet/Packets/MqttSubscribePacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttSubscribePacketProperties.cs @@ -9,6 +9,6 @@ namespace MQTTnet.Packets /// public uint SubscriptionIdentifier { get; set; } - public List UserProperties { get; set; } = new List(); + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs b/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs index 76922b94..22843f2c 100644 --- a/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttUnsubAckPacket.cs @@ -11,12 +11,12 @@ namespace MQTTnet.Packets /// /// Added in MQTT V5. /// - public MqttUnsubAckPacketProperties Properties { get; set; } = new MqttUnsubAckPacketProperties(); + public MqttUnsubAckPacketProperties Properties { get; } = new MqttUnsubAckPacketProperties(); /// /// Added in MQTT V5. /// - public List ReasonCodes { get; set; } = new List(); + public List ReasonCodes { get; } = new List(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttUnsubAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttUnsubAckPacketProperties.cs index 89ac0749..19ed103e 100644 --- a/Source/MQTTnet/Packets/MqttUnsubAckPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttUnsubAckPacketProperties.cs @@ -6,6 +6,6 @@ namespace MQTTnet.Packets { public string ReasonString { get; set; } - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs b/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs index 8518948f..64d7c007 100644 --- a/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs +++ b/Source/MQTTnet/Packets/MqttUnsubscribePacket.cs @@ -6,12 +6,12 @@ namespace MQTTnet.Packets { public ushort PacketIdentifier { get; set; } - public List TopicFilters { get; set; } = new List(); + public List TopicFilters { get; } = new List(); /// /// Added in MQTTv5. /// - public MqttUnsubscribePacketProperties Properties { get; set; } = new MqttUnsubscribePacketProperties(); + public MqttUnsubscribePacketProperties Properties { get; } = new MqttUnsubscribePacketProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttUnsubscribePacketProperties.cs b/Source/MQTTnet/Packets/MqttUnsubscribePacketProperties.cs index d98a4f99..6bb0af84 100644 --- a/Source/MQTTnet/Packets/MqttUnsubscribePacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttUnsubscribePacketProperties.cs @@ -4,6 +4,6 @@ namespace MQTTnet.Packets { public sealed class MqttUnsubscribePacketProperties { - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 3362f66f..7512a9d8 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -8,6 +8,7 @@ using MQTTnet.Client.Disconnecting; using MQTTnet.Diagnostics.Logger; using MQTTnet.Exceptions; using MQTTnet.Formatter; +using MQTTnet.Formatter.V5; using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.PacketDispatcher; @@ -69,7 +70,7 @@ namespace MQTTnet.Server.Internal public bool IsTakenOver { get; set; } public bool IsCleanDisconnect { get; private set; } - + public async Task StopAsync(MqttClientDisconnectReason reason) { IsRunning = false; @@ -163,6 +164,20 @@ namespace MQTTnet.Server.Internal { await HandleIncomingPublishPacket(publishPacket, cancellationToken).ConfigureAwait(false); } + else if (packet is MqttPubAckPacket pubAckPacket) + { + // TODO: Do anything! + } + else if (packet is MqttPubCompPacket pubCompPacket) + { + // TODO: Do anything! + } + else if (packet is MqttPubRecPacket pubRecPacket) + { + // TODO: Do anything! + // TODO: Send pub rel! + Session.EnqueuePacket(new MqttPacketBusItem(ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success))); + } else if (packet is MqttPubRelPacket pubRelPacket) { await HandleIncomingPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false); @@ -178,7 +193,8 @@ namespace MQTTnet.Server.Internal else if (packet is MqttPingReqPacket) { // See: The Server MUST send a PINGRESP packet in response to a PINGREQ packet [MQTT-3.12.4-1]. - await SendPacketAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false); + //await SendPacketAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false); + Session.EnqueuePacket(new MqttPacketBusItem(MqttPingRespPacket.Instance)); } else if (packet is MqttPingRespPacket) { @@ -216,58 +232,79 @@ namespace MQTTnet.Server.Internal async Task SendPacketsLoop(CancellationToken cancellationToken) { - MqttQueuedApplicationMessage queuedApplicationMessage = null; - MqttPublishPacket publishPacket = null; + //MqttQueuedApplicationMessage queuedApplicationMessage = null; + + MqttPacketBusItem packetBusItem = null; + //MqttPublishPacket publishPacket = null; try { while (!cancellationToken.IsCancellationRequested) { - queuedApplicationMessage = await Session.ApplicationMessagesQueue.Dequeue(cancellationToken).ConfigureAwait(false); + //queuedApplicationMessage = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false); + + packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false); // Also check the cancellation token here because the dequeue is blocking and may take some time. if (cancellationToken.IsCancellationRequested) { return; } + + try + { + await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false); + packetBusItem.MarkAsDelivered(); + + _logger.Verbose("Sent packet '{0}'.", packetBusItem.Packet); + } + catch (OperationCanceledException) + { + packetBusItem.MarkAsCancelled(); + } + catch (Exception exception) + { + packetBusItem.MarkAsFailed(exception); + } - if (queuedApplicationMessage == null) - { - continue; - } - - publishPacket = await CreatePublishPacket(queuedApplicationMessage).ConfigureAwait(false); - if (publishPacket == null) - { - continue; - } - - if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) - { - await SendPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); - } - else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) - { - using (var awaitable = _packetDispatcher.AddAwaitable(publishPacket.PacketIdentifier)) - { - await SendPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); - await awaitable.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); - } - } - else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) - { - using (var awaitableRec = _packetDispatcher.AddAwaitable(publishPacket.PacketIdentifier)) - using (var awaitableComp = _packetDispatcher.AddAwaitable(publishPacket.PacketIdentifier)) - { - await SendPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); - var pubRecPacket = await awaitableRec.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); - - var pubRelPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - await SendPacketAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); - - await awaitableComp.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); - } - } + // TODO: Move QoS 1 + 2 to other layer! + // if (queuedApplicationMessage == null) + // { + // continue; + // } + // + // publishPacket = await CreatePublishPacket(queuedApplicationMessage).ConfigureAwait(false); + // if (publishPacket == null) + // { + // continue; + // } + // + // if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) + // { + // await SendPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); + // } + // else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) + // { + // using (var awaitable = _packetDispatcher.AddAwaitable(publishPacket.PacketIdentifier)) + // { + // await SendPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); + // await awaitable.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); + // } + // } + // else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) + // { + // using (var awaitableRec = _packetDispatcher.AddAwaitable(publishPacket.PacketIdentifier)) + // using (var awaitableComp = _packetDispatcher.AddAwaitable(publishPacket.PacketIdentifier)) + // { + // await SendPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); + // var pubRecPacket = await awaitableRec.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); + // + // var pubRelPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + // await SendPacketAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); + // + // await awaitableComp.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); + // } + // } _logger.Verbose("Client '{0}': Queued application message sent.", ClientId); } @@ -290,46 +327,61 @@ namespace MQTTnet.Server.Internal _logger.Error(exception, "Client '{0}': Sending publish packet failed.", ClientId); } - if (publishPacket?.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + if (packetBusItem?.Packet is MqttPublishPacket publishPacket) { - if (queuedApplicationMessage != null) + if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - queuedApplicationMessage.IsDuplicate = true; - Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); + publishPacket.Dup = true; + Session.EnqueuePacket(new MqttPacketBusItem(publishPacket)); } + + // if (queuedApplicationMessage != null) + // { + // queuedApplicationMessage.IsDuplicate = true; + // Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); + // } } + + // if (publishPacket?.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + // { + // if (queuedApplicationMessage != null) + // { + // queuedApplicationMessage.IsDuplicate = true; + // Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); + // } + // } StopInternal(); } } - async Task CreatePublishPacket(MqttQueuedApplicationMessage queuedApplicationMessage) - { - var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage); - publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel; - - if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) - { - publishPacket.Properties.SubscriptionIdentifiers = queuedApplicationMessage.SubscriptionIdentifiers; - } - - // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9]. - publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage; - - publishPacket = await InvokeClientMessageQueueInterceptor(publishPacket, queuedApplicationMessage).ConfigureAwait(false); - if (publishPacket == null) - { - // The interceptor has decided that the message is not relevant and will be fully ignored. - return null; - } - - if (publishPacket.QualityOfServiceLevel > 0) - { - publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - } - - return publishPacket; - } + // public async Task CreatePublishPacket(MqttQueuedApplicationMessage queuedApplicationMessage) + // { + // var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage); + // publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel; + // + // if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) + // { + // publishPacket.Properties.SubscriptionIdentifiers.AddRange(queuedApplicationMessage.SubscriptionIdentifiers ?? new List()); + // } + // + // // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9]. + // publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage; + // + // publishPacket = await InvokeClientMessageQueueInterceptor(publishPacket, queuedApplicationMessage).ConfigureAwait(false); + // if (publishPacket == null) + // { + // // The interceptor has decided that the message is not relevant and will be fully ignored. + // return null; + // } + // + // if (publishPacket.QualityOfServiceLevel > 0) + // { + // publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); + // } + // + // return publishPacket; + // } void StopInternal() { @@ -347,17 +399,20 @@ namespace MQTTnet.Server.Internal var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket).ConfigureAwait(false); var subAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateSubAckPacket(subscribePacket, subscribeResult); - await SendPacketAsync(subAckPacket, cancellationToken).ConfigureAwait(false); - + //await SendPacketAsync(subAckPacket, cancellationToken).ConfigureAwait(false); + Session.EnqueuePacket(new MqttPacketBusItem(subAckPacket)); + if (subscribeResult.CloseConnection) { StopInternal(); return; } - + + var publishPacketFactory = new MqttPublishPacketFactory(); foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages) { - Session.ApplicationMessagesQueue.Enqueue(retainedApplicationMessage); + //var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(retainedApplicationMessage.ApplicationMessage); + Session.EnqueuePacket(new MqttPacketBusItem(publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage))); } } @@ -365,8 +420,9 @@ namespace MQTTnet.Server.Internal { var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket).ConfigureAwait(false); var unsubAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, unsubscribeResult); - - await SendPacketAsync(unsubAckPacket, cancellationToken).ConfigureAwait(false); + + Session.EnqueuePacket(new MqttPacketBusItem(unsubAckPacket)); + //await SendPacketAsync(unsubAckPacket, cancellationToken).ConfigureAwait(false); if (unsubscribeResult.CloseConnection) { @@ -378,30 +434,42 @@ namespace MQTTnet.Server.Internal { HandleTopicAlias(publishPacket); - var applicationMessage = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket); + var applicationMessageFactory = new MqttApplicationMessageFactory(); + var applicationMessage = applicationMessageFactory.Create(publishPacket); + _sessionsManager.DispatchApplicationMessage(applicationMessage, this); + // TODO: Check! switch (publishPacket.QualityOfServiceLevel) { case MqttQualityOfServiceLevel.AtMostOnce: { - return PlatformAbstractionLayer.CompletedTask; + // Do nothing since QoS 0 has no ack at all! + break; } case MqttQualityOfServiceLevel.AtLeastOnce: { var pubAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); - return SendPacketAsync(pubAckPacket, cancellationToken); + Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket)); + //return SendPacketAsync(pubAckPacket, cancellationToken); + break; } case MqttQualityOfServiceLevel.ExactlyOnce: { + // TODO: Migrate to packet bus. var pubRecPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); - return SendPacketAsync(pubRecPacket, cancellationToken); + Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket)); + + //return SendPacketAsync(pubRecPacket, cancellationToken); + break; } default: { throw new MqttCommunicationException("Received a not supported QoS level."); } } + + return PlatformAbstractionLayer.CompletedTask; } void HandleTopicAlias(MqttPublishPacket publishPacket) @@ -442,7 +510,8 @@ namespace MQTTnet.Server.Internal ReasonString = reason.ToString() }; - var disconnectPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(disconnectOptions); + var disconnectPacketFactory = new MqttDisconnectPacketFactory(); + var disconnectPacket = disconnectPacketFactory.Create(disconnectOptions); using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout)) { diff --git a/Source/MQTTnet/Server/Internal/MqttClientSession.cs b/Source/MQTTnet/Server/Internal/MqttClientSession.cs index 1b48f5b6..8a189c51 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSession.cs @@ -1,11 +1,18 @@ using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; +using MQTTnet.Client; +using MQTTnet.Internal; +using MQTTnet.Packets; namespace MQTTnet.Server.Internal { - public sealed class MqttClientSession + public sealed class MqttClientSession : IDisposable { + readonly MqttPacketBus _packetBus = new MqttPacketBus(); + + readonly IMqttServerOptions _serverOptions; readonly MqttClientSessionsManager _clientSessionsManager; public MqttClientSession(string clientId, @@ -15,18 +22,21 @@ namespace MQTTnet.Server.Internal IMqttRetainedMessagesManager retainedMessagesManager, MqttClientSessionsManager clientSessionsManager) { + _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager)); ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); Items = items ?? throw new ArgumentNullException(nameof(items)); SubscriptionsManager = new MqttClientSubscriptionsManager(this, serverOptions, eventDispatcher, retainedMessagesManager); - ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions); + //ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions); } public event EventHandler Deleted; public string ClientId { get; } + public MqttPacketIdentifierProvider PacketIdentifierProvider { get; } = new MqttPacketIdentifierProvider(); + public DateTime CreatedTimestamp { get; } = DateTime.UtcNow; public bool IsCleanSession { get; set; } = true; @@ -35,10 +45,39 @@ namespace MQTTnet.Server.Internal public MqttClientSubscriptionsManager SubscriptionsManager { get; } - public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; } + //public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; } public IDictionary Items { get; } + public void EnqueuePacket(MqttPacketBusItem packet) + { + if (packet == null) throw new ArgumentNullException(nameof(packet)); + + if (_packetBus.PacketsCount >= _serverOptions.MaxPendingMessagesPerClient) + { + // TODO: Log! + return; + } + + if (packet.Packet is MqttPublishPacket) + { + _packetBus.Enqueue(packet, MqttPacketBusPartition.Data); + } + else if (packet.Packet is MqttPingReqPacket || packet.Packet is MqttPingRespPacket) + { + _packetBus.Enqueue(packet, MqttPacketBusPartition.Health); + } + else + { + _packetBus.Enqueue(packet, MqttPacketBusPartition.Control); + } + } + + public Task DequeuePacketAsync(CancellationToken cancellationToken) + { + return _packetBus.DequeueAsync(cancellationToken); + } + public Task DeleteAsync() { return _clientSessionsManager.DeleteSessionAsync(ClientId); @@ -48,5 +87,10 @@ namespace MQTTnet.Server.Internal { Deleted?.Invoke(this, EventArgs.Empty); } + + public void Dispose() + { + _packetBus?.Dispose(); + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index f7f41798..75ee9143 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -9,6 +9,8 @@ using MQTTnet.Client.Disconnecting; using MQTTnet.Diagnostics.Logger; using MQTTnet.Exceptions; using MQTTnet.Formatter; +using MQTTnet.Formatter.V3; +using MQTTnet.Formatter.V5; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -229,10 +231,8 @@ namespace MQTTnet.Server.Internal if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - var fakeSubscribePacket = new MqttSubscribePacket - { - TopicFilters = topicFilters.ToList() - }; + var fakeSubscribePacket = new MqttSubscribePacket(); + fakeSubscribePacket.TopicFilters.AddRange(topicFilters); var clientSession = GetClientSession(clientId); @@ -240,7 +240,12 @@ namespace MQTTnet.Server.Internal foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages) { - clientSession.ApplicationMessagesQueue.Enqueue(retainedApplicationMessage); + //clientSession.ApplicationMessagesQueue.Enqueue(retainedApplicationMessage); + + // TODO: Review! + var publishPacketFactory = new MqttPublishPacketFactory(); + var publishPacket = publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage); + clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket)); } } @@ -249,10 +254,8 @@ namespace MQTTnet.Server.Internal if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - var fakeUnsubscribePacket = new MqttUnsubscribePacket - { - TopicFilters = topicFilters.ToList() - }; + var fakeUnsubscribePacket = new MqttUnsubscribePacket(); + fakeUnsubscribePacket.TopicFilters.AddRange(topicFilters); return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket); } @@ -273,12 +276,28 @@ namespace MQTTnet.Server.Internal _clientSessions.Remove(clientId); } - if (connection != null) + try { - await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); + if (connection != null) + { + await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); + } + } + catch (Exception exception) + { + _logger.Error(exception, $"Error while deleting session '{clientId}'."); } - session?.OnDeleted(); + try + { + session?.OnDeleted(); + } + catch (Exception exception) + { + _logger.Error(exception, $"Error while executing session deleted event for session '{clientId}'."); + } + + session?.Dispose(); _logger.Verbose("Session for client '{0}' deleted.", clientId); } @@ -320,6 +339,10 @@ namespace MQTTnet.Server.Internal { return; } + // catch (ThreadAbortException) + // { + // return; + // } catch (ObjectDisposedException) { return; @@ -380,21 +403,39 @@ namespace MQTTnet.Server.Internal _logger.Verbose("Client '{0}': Queued application message with topic '{1}'.", clientSession.ClientId, applicationMessage.Topic); - var queuedApplicationMessage = new MqttQueuedApplicationMessage + // var queuedApplicationMessage = new MqttQueuedApplicationMessage + // { + // ApplicationMessage = applicationMessage, + // SubscriptionQualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel, + // SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers + // }; + + + + var publishPacketFactory = new MqttPublishPacketFactory(); + var publishPacket = publishPacketFactory.Create(applicationMessage); + + publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel; + publishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers); + //clientSession.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); + + if (publishPacket.QualityOfServiceLevel > 0) { - ApplicationMessage = applicationMessage, - SubscriptionQualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel, - SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers - }; + publishPacket.PacketIdentifier = clientSession.PacketIdentifierProvider.GetNextPacketIdentifier(); + } if (checkSubscriptionsResult.RetainAsPublished) { // Transfer the original retain state from the publisher. // This is a MQTTv5 feature. - queuedApplicationMessage.IsRetainedMessage = applicationMessage.Retain; + publishPacket.Retain = applicationMessage.Retain; } - - clientSession.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); + else + { + publishPacket.Retain = false; + } + + clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket)); deliveryCount++; } @@ -415,6 +456,34 @@ namespace MQTTnet.Server.Internal _logger.Error(exception, "Unhandled exception while processing next queued application message."); } } + + // public async Task CreatePublishPacket(MqttQueuedApplicationMessage queuedApplicationMessage) + // { + // var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage); + // publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel; + // + // if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) + // { + // publishPacket.Properties.SubscriptionIdentifiers.AddRange(queuedApplicationMessage.SubscriptionIdentifiers ?? new List()); + // } + // + // // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9]. + // publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage; + // + // publishPacket = await InvokeClientMessageQueueInterceptor(publishPacket, queuedApplicationMessage).ConfigureAwait(false); + // if (publishPacket == null) + // { + // // The interceptor has decided that the message is not relevant and will be fully ignored. + // return null; + // } + // + // if (publishPacket.QualityOfServiceLevel > 0) + // { + // publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); + // } + // + // return publishPacket; + // } async Task ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) { diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index 0ee53a76..f13b6826 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -1,7 +1,10 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; +using MQTTnet.Formatter; +using MQTTnet.Formatter.V5; using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Server.Internal; namespace MQTTnet.Server.Status @@ -23,7 +26,8 @@ namespace MQTTnet.Server.Status public string ClientId => _session.ClientId; - public long PendingApplicationMessagesCount => _session.ApplicationMessagesQueue.Count; + // TODO: Fix! + public long PendingApplicationMessagesCount => 0; // _session.ApplicationMessagesQueue.Count; public DateTime CreatedTimestamp => _session.CreatedTimestamp; @@ -33,15 +37,18 @@ namespace MQTTnet.Server.Status { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - _session.ApplicationMessagesQueue.Enqueue(new MqttQueuedApplicationMessage - { - ApplicationMessage = applicationMessage, - IsDuplicate = false, - IsRetainedMessage = false, - SubscriptionQualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, - SenderClientId = null - }); + // _session.ApplicationMessagesQueue.Enqueue(new MqttQueuedApplicationMessage + // { + // ApplicationMessage = applicationMessage, + // IsDuplicate = false, + // IsRetainedMessage = false, + // SubscriptionQualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, + // SenderClientId = null + // }); + var publishPacketFactory = new MqttPublishPacketFactory(); + _session.EnqueuePacket(new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage))); + return PlatformAbstractionLayer.CompletedTask; } @@ -49,12 +56,17 @@ namespace MQTTnet.Server.Status { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - throw new NotImplementedException(); + var publishPacketFactory = new MqttPublishPacketFactory(); + var packetBusItem = new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage)); + _session.EnqueuePacket(packetBusItem); + + return packetBusItem.WaitForDeliveryAsync(); } public Task ClearApplicationMessagesQueueAsync() { - _session.ApplicationMessagesQueue.Clear(); + // TODO: Fix! + //_session.ApplicationMessagesQueue.Clear(); return Task.FromResult(0); } diff --git a/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs index 79d1855a..77d7d026 100644 --- a/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs @@ -143,7 +143,7 @@ namespace MQTTnet.Tests.Client await client.SendAsync(new MqttSubscribePacket { PacketIdentifier = 1, - TopicFilters = new List + TopicFilters = { new MqttTopicFilter { diff --git a/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs index 0e188f06..8ee99b8a 100644 --- a/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs @@ -10,6 +10,7 @@ using MQTTnet.Client; using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Options; +using MQTTnet.Client.Publishing; using MQTTnet.Client.Receiving; using MQTTnet.Client.Subscribing; using MQTTnet.Exceptions; @@ -870,5 +871,83 @@ namespace MQTTnet.Tests.Client Assert.IsFalse(disconnectedFired); } } + + [DataTestMethod] + [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)] + [DataRow(MqttQualityOfServiceLevel.AtMostOnce)] + [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)] + public async Task Concurrent_Processing(MqttQualityOfServiceLevel qos) + { + long concurrency = 0; + bool success = false; + + using (var testEnvironment = new TestEnvironment(TestContext)) + { + await testEnvironment.StartServer(); + var publisher = await testEnvironment.ConnectClient(); + var subscriber = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId(qos.ToString())); + await subscriber.SubscribeAsync("#", qos); + + subscriber.UseApplicationMessageReceivedHandler(c => + { + c.AutoAcknowledge = false; + + async Task InvokeInternal() + { + if (Interlocked.Increment(ref concurrency) > 1) success = true; + await Task.Delay(100); + Interlocked.Decrement(ref concurrency); + } + + _ = InvokeInternal(); + return Task.CompletedTask; + }); + + var publishes = Task.WhenAll( + publisher.PublishAsync("a", null, qos), + publisher.PublishAsync("b", null, qos) + ); + + await Task.Delay(200); + + await publishes; + Assert.IsTrue(success); + } + } + + [DataTestMethod] + [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)] + [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)] + public async Task Retry_If_Not_PubAck(MqttQualityOfServiceLevel qos) + { + long count = 0; + using (var testEnvironment = new TestEnvironment(TestContext)) + { + await testEnvironment.StartServer(); + var publisher = await testEnvironment.ConnectClient(); + var subscriber = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId(qos.ToString())); + await subscriber.SubscribeAsync("#", qos); + + subscriber.UseApplicationMessageReceivedHandler(c => + { + c.AutoAcknowledge = false; + ++count; + Console.WriteLine("process"); + return Task.CompletedTask; + }); + + var pub = publisher.PublishAsync("a", null, qos); + + await Task.Delay(100); + await subscriber.DisconnectAsync(); + await subscriber.ReconnectAsync(); + await Task.Delay(100); + + var res = await pub; + + Assert.AreEqual(MqttClientPublishReasonCode.Success, res.ReasonCode); + Assert.AreEqual(2, count); + } + } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index 26c23239..511f3dae 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs @@ -1,12 +1,10 @@ using System; -using System.Collections.Generic; using System.IO; using System.Linq; using System.Text; using System.Threading; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Adapter; -using MQTTnet.Diagnostics; using MQTTnet.Diagnostics.Logger; using MQTTnet.Exceptions; using MQTTnet.Formatter; @@ -278,12 +276,6 @@ namespace MQTTnet.Tests [TestMethod] public void SerializeV500_MqttPublishPacket() { - var prop = new MqttPublishPacketProperties { UserProperties = new List() }; - - prop.ResponseTopic = "/Response"; - - prop.UserProperties.Add(new MqttUserProperty("Foo", "Bar")); - var p = new MqttPublishPacket { PacketIdentifier = 123, @@ -292,12 +284,17 @@ namespace MQTTnet.Tests Payload = Encoding.ASCII.GetBytes("HELLO"), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, Topic = "A/B/C", - Properties = prop + Properties = + { + ResponseTopic = "/Response" + } }; + p.Properties.UserProperties.Add(new MqttUserProperty("Foo", "Bar")); + var deserialized = Roundtrip(p, MqttProtocolVersion.V500); - Assert.AreEqual(prop.ResponseTopic, deserialized.Properties.ResponseTopic); + Assert.AreEqual(p.Properties.ResponseTopic, deserialized.Properties.ResponseTopic); Assert.IsTrue(deserialized.Properties.UserProperties.Any(x => x.Name == "Foo")); } @@ -314,7 +311,7 @@ namespace MQTTnet.Tests .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce) .Build(); - var p = new MqttV500DataConverter().CreatePublishPacket(req); + var p = new MqttPublishPacketFactory().Create(req); var deserialized = Roundtrip(p, MqttProtocolVersion.V500); diff --git a/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs index 66365984..4cbb8585 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs @@ -21,7 +21,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public Task Connect_With_Client_Id() { - return Connect_With_Client_Id("Connect_With_Client_Idtest_456", null, "test_456", null); + return Connect_With_Client_Id("Connect_With_Client_Id_test_456", null, "test_456", null); } async Task Connect_With_Client_Id(string expectedClientId, string expectedReturnedClientId, string usedClientId, string assignedClientId) diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index 28e8e1a0..5ba35b57 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -955,23 +955,23 @@ namespace MQTTnet.Tests.Server { using (var testEnvironment = CreateTestEnvironment()) { - var receivedMessagesCount = 0; - await testEnvironment.StartServer(); var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("receiver")); - c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); + var c1MessageHandler = testEnvironment.CreateApplicationMessageHandler(c1); await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build()); var c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("sender")); await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); + await Task.Delay(500); + await c2.DisconnectAsync().ConfigureAwait(false); await Task.Delay(500); await c1.UnsubscribeAsync(topicFilter); await Task.Delay(500); - Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount); + Assert.AreEqual(expectedReceivedMessagesCount, c1MessageHandler.ReceivedApplicationMessages.Count); } }