diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 70594cac..bc0b1682 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -680,7 +680,8 @@ namespace MQTTnet.Client { if (!eventArgs.ProcessingFailed) { - var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(eventArgs.PublishPacket, eventArgs.ReasonCode); + var pubAckPacketFactory = new MqttPubAckPacketFactory(); + var pubAckPacket = pubAckPacketFactory.Create(eventArgs); return SendAsync(pubAckPacket, cancellationToken); } } @@ -688,7 +689,8 @@ namespace MQTTnet.Client { if (!eventArgs.ProcessingFailed) { - var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(eventArgs.PublishPacket, eventArgs.ReasonCode); + var pubRecPacketFactory = new MqttPubRecPacketFactory(); + var pubRecPacket = pubRecPacketFactory.Create(eventArgs); return SendAsync(pubRecPacket, cancellationToken); } } diff --git a/Source/MQTTnet/Formatter/IMqttDataConverter.cs b/Source/MQTTnet/Formatter/IMqttDataConverter.cs index d8f9980e..4fea2562 100644 --- a/Source/MQTTnet/Formatter/IMqttDataConverter.cs +++ b/Source/MQTTnet/Formatter/IMqttDataConverter.cs @@ -3,7 +3,6 @@ using MQTTnet.Client.Publishing; using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; using MQTTnet.Packets; -using MQTTnet.Server; using MQTTnet.Server.Internal; namespace MQTTnet.Formatter @@ -20,12 +19,6 @@ namespace MQTTnet.Formatter MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket); - MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext); - - MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode); - - MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode); - MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode); MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode); diff --git a/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs new file mode 100644 index 00000000..fd05b369 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs @@ -0,0 +1,41 @@ +using System; +using MQTTnet.Packets; +using MQTTnet.Server; + +namespace MQTTnet.Formatter +{ + public sealed class MqttConnAckPacketFactory + { + public MqttConnAckPacket Create(MqttConnectionValidatorContext connectionValidatorContext) + { + if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext)); + + var connAckPacket = new MqttConnAckPacket + { + ReturnCode = MqttConnectReasonCodeConverter.ToConnectReturnCode(connectionValidatorContext.ReasonCode), + ReasonCode = connectionValidatorContext.ReasonCode, + Properties = + { + RetainAvailable = true, + SubscriptionIdentifiersAvailable = true, + SharedSubscriptionAvailable = false, + TopicAliasMaximum = ushort.MaxValue, + WildcardSubscriptionAvailable = true, + + AuthenticationMethod = connectionValidatorContext.AuthenticationMethod, + AuthenticationData = connectionValidatorContext.ResponseAuthenticationData, + AssignedClientIdentifier = connectionValidatorContext.AssignedClientIdentifier, + ReasonString = connectionValidatorContext.ReasonString, + ServerReference = connectionValidatorContext.ServerReference + } + }; + + if (connectionValidatorContext.ResponseUserProperties != null) + { + connAckPacket.Properties.UserProperties.AddRange(connectionValidatorContext.ResponseUserProperties); + } + + return connAckPacket; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Protocol/MqttConnectReasonCodeConverter.cs b/Source/MQTTnet/Formatter/MqttConnectReasonCodeConverter.cs similarity index 91% rename from Source/MQTTnet/Protocol/MqttConnectReasonCodeConverter.cs rename to Source/MQTTnet/Formatter/MqttConnectReasonCodeConverter.cs index d195b687..54995b20 100644 --- a/Source/MQTTnet/Protocol/MqttConnectReasonCodeConverter.cs +++ b/Source/MQTTnet/Formatter/MqttConnectReasonCodeConverter.cs @@ -1,10 +1,11 @@ using MQTTnet.Exceptions; +using MQTTnet.Protocol; -namespace MQTTnet.Protocol +namespace MQTTnet.Formatter { - public class MqttConnectReasonCodeConverter + public static class MqttConnectReasonCodeConverter { - public MqttConnectReturnCode ToConnectReturnCode(MqttConnectReasonCode reasonCode) + public static MqttConnectReturnCode ToConnectReturnCode(MqttConnectReasonCode reasonCode) { switch (reasonCode) { @@ -47,7 +48,7 @@ namespace MQTTnet.Protocol } } - public MqttConnectReasonCode ToConnectReasonCode(MqttConnectReturnCode returnCode) + public static MqttConnectReasonCode ToConnectReasonCode(MqttConnectReturnCode returnCode) { switch (returnCode) { diff --git a/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs new file mode 100644 index 00000000..ffc0f425 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs @@ -0,0 +1,32 @@ +using System; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter +{ + public sealed class MqttPubAckPacketFactory + { + public MqttPubAckPacket Create(MqttApplicationMessageReceivedEventArgs applicationMessageReceivedEventArgs) + { + if (applicationMessageReceivedEventArgs == null) throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs)); + + var pubAckPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode); + pubAckPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties); + + return pubAckPacket; + } + + public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode) + { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + + var pubAckPacket = new MqttPubAckPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = (MqttPubAckReasonCode) (int) applicationMessageReceivedReasonCode + }; + + return pubAckPacket; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs new file mode 100644 index 00000000..9fb369d0 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs @@ -0,0 +1,35 @@ + + + +using System; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter +{ + public sealed class MqttPubRecPacketFactory + { + public MqttPubRecPacket Create(MqttApplicationMessageReceivedEventArgs applicationMessageReceivedEventArgs) + { + if (applicationMessageReceivedEventArgs == null) throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs)); + + var pubRecPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode); + pubRecPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties); + + return pubRecPacket; + } + + public MqttPubRecPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode) + { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + + var pubRecPacket = new MqttPubRecPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = (MqttPubRecReasonCode) (int) applicationMessageReceivedReasonCode + }; + + return pubRecPacket; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs index db7b7406..6af35a6e 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs @@ -14,26 +14,6 @@ namespace MQTTnet.Formatter.V3 { public class MqttV310DataConverter : IMqttDataConverter { - public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - return new MqttPubAckPacket - { - PacketIdentifier = publishPacket.PacketIdentifier - }; - } - - public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - return new MqttPubRecPacket - { - PacketIdentifier = publishPacket.PacketIdentifier - }; - } - public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket)); @@ -110,16 +90,6 @@ namespace MQTTnet.Formatter.V3 }; } - public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext) - { - if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext)); - - return new MqttConnAckPacket - { - ReturnCode = new MqttConnectReasonCodeConverter().ToConnectReturnCode(connectionValidatorContext.ReasonCode) - }; - } - public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index 410cc869..79e97168 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -5,7 +5,6 @@ using MQTTnet.Client.Unsubscribing; using MQTTnet.Exceptions; using MQTTnet.Packets; using MQTTnet.Protocol; -using MQTTnet.Server; using System; using System.Collections.Generic; using System.Linq; @@ -15,30 +14,6 @@ namespace MQTTnet.Formatter.V5 { public sealed class MqttV500DataConverter : IMqttDataConverter { - public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, - MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - return new MqttPubAckPacket - { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = (MqttPubAckReasonCode) (int) reasonCode - }; - } - - public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, - MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - return new MqttPubRecPacket - { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = (MqttPubRecReasonCode) (int) reasonCode - }; - } - public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { @@ -91,37 +66,6 @@ namespace MQTTnet.Formatter.V5 }; } - public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext) - { - if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext)); - - var connAckPacket = new MqttConnAckPacket - { - ReasonCode = connectionValidatorContext.ReasonCode, - Properties = - { - RetainAvailable = true, - SubscriptionIdentifiersAvailable = true, - SharedSubscriptionAvailable = false, - TopicAliasMaximum = ushort.MaxValue, - WildcardSubscriptionAvailable = true, - - AuthenticationMethod = connectionValidatorContext.AuthenticationMethod, - AuthenticationData = connectionValidatorContext.ResponseAuthenticationData, - AssignedClientIdentifier = connectionValidatorContext.AssignedClientIdentifier, - ReasonString = connectionValidatorContext.ReasonString, - ServerReference = connectionValidatorContext.ServerReference - } - }; - - if (connectionValidatorContext.ResponseUserProperties != null) - { - connAckPacket.Properties.UserProperties.AddRange(connectionValidatorContext.ResponseUserProperties); - } - - return connAckPacket; - } - public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) { diff --git a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs index ea54dd5c..c3578a61 100644 --- a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs +++ b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using MQTTnet.Packets; @@ -36,7 +37,12 @@ namespace MQTTnet public bool ProcessingFailed { get; set; } public MqttApplicationMessageReceivedReasonCode ReasonCode { get; set; } = MqttApplicationMessageReceivedReasonCode.Success; - + + /// + /// Gets or sets the user properties which being sent to the server in the ACK packet etc. + /// + public List ResponseUserProperties { get; } = new List(); + /// /// Gets or sets whether this message was handled. /// This value can be used in user code for custom control flow. diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 7512a9d8..6541891d 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -449,7 +449,8 @@ namespace MQTTnet.Server.Internal } case MqttQualityOfServiceLevel.AtLeastOnce: { - var pubAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubAckPacketFactory = new MqttPubAckPacketFactory(); + var pubAckPacket = pubAckPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket)); //return SendPacketAsync(pubAckPacket, cancellationToken); break; @@ -457,7 +458,8 @@ namespace MQTTnet.Server.Internal case MqttQualityOfServiceLevel.ExactlyOnce: { // TODO: Migrate to packet bus. - var pubRecPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubRecPacketFactory = new MqttPubRecPacketFactory(); + var pubRecPacket = pubRecPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket)); //return SendPacketAsync(pubRecPacket, cancellationToken); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 75ee9143..e6bb2d0b 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -9,8 +9,6 @@ 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; @@ -28,6 +26,8 @@ namespace MQTTnet.Server.Internal readonly IDictionary _serverSessionItems = new ConcurrentDictionary(); + readonly MqttConnAckPacketFactory _connAckPacketFactory = new MqttConnAckPacketFactory(); + readonly MqttServerEventDispatcher _eventDispatcher; readonly IMqttRetainedMessagesManager _retainedMessagesManager; @@ -102,14 +102,14 @@ namespace MQTTnet.Server.Internal if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success) { // Send failure response here without preparing a session! - connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext); + connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext); await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false); return; } clientConnection = await CreateClientConnection(connectPacket, channelAdapter, connectionValidatorContext.SessionItems).ConfigureAwait(false); - connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext); + connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext); await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false); await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket, channelAdapter).ConfigureAwait(false); diff --git a/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs b/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs index c0eb8a21..c48b35ab 100644 --- a/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs +++ b/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs @@ -136,8 +136,8 @@ namespace MQTTnet.Server [Obsolete("Use ReasonCode instead. It is MQTTv5 only but will be converted to a valid ReturnCode.")] public MqttConnectReturnCode ReturnCode { - get => new MqttConnectReasonCodeConverter().ToConnectReturnCode(ReasonCode); - set => ReasonCode = new MqttConnectReasonCodeConverter().ToConnectReasonCode(value); + get => MqttConnectReasonCodeConverter.ToConnectReturnCode(ReasonCode); + set => ReasonCode = MqttConnectReasonCodeConverter.ToConnectReasonCode(value); } ///