diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 4ffbc417..47efd2ed 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -16,6 +16,8 @@ * [Core] Removed global logger and refactored logging (BREAKING CHANGE!). * [Client] Renamed _MqttClientConnectResult_ to _MqttClientConnectResult_ (BREAKING CHANGE!). * [Client] The OS will now choose the best TLS version to use. It is no longer fixed to 1.3 etc. (thanks to @patagonaa, #1271). +* [Client] Exposed user properties and reason string in subscribe result. +* [Client] Exposed user properties and reason string in unsubscribe result. * [ManagedClient] Extended ReconnectAsync (thanks to @nvsnkv, #1202). * [ManagedClient] Improved Amazon AWS support (thanks to @scottbrogden-iheartmedia, #1209). * [ManagedClient] Fixed bug that allowed invalid subscriptions (Thanks to @marcelwinh). diff --git a/Source/MQTTnet/Client/Connecting/MqttClientConnectResultFactory.cs b/Source/MQTTnet/Client/Connecting/MqttClientConnectResultFactory.cs new file mode 100644 index 00000000..0f1b413c --- /dev/null +++ b/Source/MQTTnet/Client/Connecting/MqttClientConnectResultFactory.cs @@ -0,0 +1,103 @@ +using System; +using MQTTnet.Exceptions; +using MQTTnet.Formatter; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Client.Connecting +{ + public sealed class MqttClientConnectResultFactory + { + public MqttClientConnectResult Create(MqttConnAckPacket connAckPacket, MqttProtocolVersion protocolVersion) + { + if (connAckPacket == null) throw new ArgumentNullException(nameof(connAckPacket)); + + if (protocolVersion == MqttProtocolVersion.V500) + { + return CreateForMqtt500(connAckPacket); + } + + return CreateForMqtt311(connAckPacket); + } + + static MqttClientConnectResult CreateForMqtt500(MqttConnAckPacket connAckPacket) + { + if (connAckPacket == null) throw new ArgumentNullException(nameof(connAckPacket)); + + return new MqttClientConnectResult + { + IsSessionPresent = connAckPacket.IsSessionPresent, + ResultCode = (MqttClientConnectResultCode) (int) connAckPacket.ReasonCode, + WildcardSubscriptionAvailable = connAckPacket.Properties.WildcardSubscriptionAvailable, + RetainAvailable = connAckPacket.Properties.RetainAvailable, + AssignedClientIdentifier = connAckPacket.Properties.AssignedClientIdentifier, + AuthenticationMethod = connAckPacket.Properties.AuthenticationMethod, + AuthenticationData = connAckPacket.Properties.AuthenticationData, + MaximumPacketSize = connAckPacket.Properties.MaximumPacketSize, + ReasonString = connAckPacket.Properties.ReasonString, + ReceiveMaximum = connAckPacket.Properties.ReceiveMaximum, + MaximumQoS = connAckPacket.Properties.MaximumQoS ?? MqttQualityOfServiceLevel.ExactlyOnce, + ResponseInformation = connAckPacket.Properties.ResponseInformation, + TopicAliasMaximum = connAckPacket.Properties.TopicAliasMaximum, + ServerReference = connAckPacket.Properties.ServerReference, + ServerKeepAlive = connAckPacket.Properties.ServerKeepAlive, + SessionExpiryInterval = connAckPacket.Properties?.SessionExpiryInterval, + SubscriptionIdentifiersAvailable = connAckPacket.Properties.SubscriptionIdentifiersAvailable, + SharedSubscriptionAvailable = connAckPacket.Properties.SharedSubscriptionAvailable, + UserProperties = connAckPacket.Properties.UserProperties + }; + } + + static MqttClientConnectResult CreateForMqtt311(MqttConnAckPacket connAckPacket) + { + if (connAckPacket == null) throw new ArgumentNullException(nameof(connAckPacket)); + + return new MqttClientConnectResult + { + RetainAvailable = true, // Always true because v3.1.1 does not have a way to "disable" that feature. + WildcardSubscriptionAvailable = true, // Always true because v3.1.1 does not have a way to "disable" that feature. + IsSessionPresent = connAckPacket.IsSessionPresent, + ResultCode = ConvertReturnCodeToResultCode(connAckPacket.ReturnCode) + }; + } + + static MqttClientConnectResultCode ConvertReturnCodeToResultCode(MqttConnectReturnCode connectReturnCode) + { + switch (connectReturnCode) + { + case MqttConnectReturnCode.ConnectionAccepted: + { + return MqttClientConnectResultCode.Success; + } + + case MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion: + { + return MqttClientConnectResultCode.UnsupportedProtocolVersion; + } + + case MqttConnectReturnCode.ConnectionRefusedNotAuthorized: + { + return MqttClientConnectResultCode.NotAuthorized; + } + + case MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword: + { + return MqttClientConnectResultCode.BadUserNameOrPassword; + } + + case MqttConnectReturnCode.ConnectionRefusedIdentifierRejected: + { + return MqttClientConnectResultCode.ClientIdentifierNotValid; + } + + case MqttConnectReturnCode.ConnectionRefusedServerUnavailable: + { + return MqttClientConnectResultCode.ServerUnavailable; + } + + default: + throw new MqttProtocolViolationException("Received unexpected return code."); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index bc0b1682..8a71eab0 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -92,10 +92,10 @@ namespace MQTTnet.Client var adapter = _adapterFactory.CreateClientAdapter(options); _adapter = adapter; - using (var combined = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken)) + using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken)) { _logger.Verbose("Trying to connect with server '{0}' (Timeout={1}).", options.ChannelOptions, options.CommunicationTimeout); - await adapter.ConnectAsync(options.CommunicationTimeout, combined.Token).ConfigureAwait(false); + await adapter.ConnectAsync(options.CommunicationTimeout, effectiveCancellationToken.Token).ConfigureAwait(false); _logger.Verbose("Connection with server established."); _publishPacketReceiverQueue = new AsyncQueue(); @@ -103,7 +103,7 @@ namespace MQTTnet.Client _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken); - connectResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false); + connectResult = await AuthenticateAsync(options, effectiveCancellationToken.Token).ConfigureAwait(false); } _lastPacketSentTimestamp = DateTime.UtcNow; @@ -212,7 +212,9 @@ namespace MQTTnet.Client subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); - return _adapter.PacketFormatterAdapter.DataConverter.CreateClientSubscribeResult(subscribePacket, subAckPacket); + + var clientSubscribeResultFactory = new MqttClientSubscribeResultFactory(); + return clientSubscribeResultFactory.Create(subscribePacket, subAckPacket); } public async Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken) @@ -227,7 +229,9 @@ namespace MQTTnet.Client unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); - return _adapter.PacketFormatterAdapter.DataConverter.CreateClientUnsubscribeResult(unsubscribePacket, unsubAckPacket); + + var clientUnsubscribeResultFactory = new MqttClientUnsubscribeResultFactory(); + return clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket); } public Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) @@ -293,17 +297,19 @@ namespace MQTTnet.Client base.Dispose(disposing); } - async Task AuthenticateAsync(IMqttChannelAdapter channelAdapter, MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) + async Task AuthenticateAsync(IMqttClientOptions options, CancellationToken cancellationToken) { MqttClientConnectResult result; try { var connectPacketFactory = new MqttConnectPacketFactory(); - var connectPacket = connectPacketFactory.Create(Options); + var connectPacket = connectPacketFactory.Create(options); var connAckPacket = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false); - result = channelAdapter.PacketFormatterAdapter.DataConverter.CreateClientConnectResult(connAckPacket); + + var clientConnectResultFactory = new MqttClientConnectResultFactory(); + result = clientConnectResultFactory.Create(connAckPacket, options.ProtocolVersion); } catch (Exception exception) { @@ -708,7 +714,8 @@ namespace MQTTnet.Client { // The packet is unknown. Probably due to a restart of the client. // So wen send this to the server to trigger a full resend of the message. - var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); + var pubRelPacketFactory = new MqttPubRelPacketFactory(); + var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); return SendAsync(pubRelPacket, cancellationToken); } @@ -717,7 +724,8 @@ namespace MQTTnet.Client Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { - var pubCompPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubCompPacketFactory = new MqttPubCompPacketFactory(); + var pubCompPacket =pubCompPacketFactory.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); return SendAsync(pubCompPacket, cancellationToken); } @@ -746,7 +754,8 @@ namespace MQTTnet.Client { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false); - return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(null); + + return new MqttClientPublishResultFactory().Create(null); } async Task PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -754,8 +763,9 @@ namespace MQTTnet.Client publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var pubAckPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - - return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubAckPacket); + + var clientPublishResultFactory = new MqttClientPublishResultFactory(); + return clientPublishResultFactory.Create(pubAckPacket); } async Task PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -764,11 +774,13 @@ namespace MQTTnet.Client var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubRelPacketFactory = new MqttPubRelPacketFactory(); + var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); - return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubRecPacket, pubCompPacket); + var clientPublishResultFactory = new MqttClientPublishResultFactory(); + return clientPublishResultFactory.Create(pubRecPacket, pubCompPacket); } async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishReasonCode.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishReasonCode.cs index 7b485c6b..dddb8d07 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishReasonCode.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishReasonCode.cs @@ -3,6 +3,7 @@ public enum MqttClientPublishReasonCode { Success = 0, + NoMatchingSubscribers = 16, UnspecifiedError = 128, ImplementationSpecificError = 131, diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs index 429f130a..a5d9e6cd 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs @@ -4,7 +4,7 @@ using MQTTnet.Packets; namespace MQTTnet.Client.Publishing { - public class MqttClientPublishResult + public sealed class MqttClientPublishResult { public ushort? PacketIdentifier { get; set; } @@ -27,6 +27,6 @@ namespace MQTTnet.Client.Publishing /// The feature is very similar to the HTTP header concept. /// Hint: MQTT 5 feature only. /// - public List UserProperties { get; set; } + public List UserProperties { get; } = new List(); } } diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs new file mode 100644 index 00000000..f97e3fe4 --- /dev/null +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs @@ -0,0 +1,74 @@ +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Client.Publishing +{ + public sealed class MqttClientPublishResultFactory + { + public MqttClientPublishResult Create(MqttPubAckPacket pubAckPacket) + { + var result = new MqttClientPublishResult + { + ReasonCode = MqttClientPublishReasonCode.Success, + }; + + if (pubAckPacket != null) + { + result.ReasonString = pubAckPacket.Properties.ReasonString; + result.UserProperties.AddRange(pubAckPacket.Properties.UserProperties); + + // 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; + + result.PacketIdentifier = pubAckPacket.PacketIdentifier; + } + + return result; + } + + public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) + { + if (pubRecPacket == null || pubCompPacket == null) + { + return new MqttClientPublishResult + { + ReasonCode = MqttClientPublishReasonCode.UnspecifiedError + }; + } + + MqttClientPublishResult result; + + // The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC. + if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound) + { + result = new MqttClientPublishResult + { + PacketIdentifier = pubCompPacket.PacketIdentifier, + ReasonCode = MqttClientPublishReasonCode.UnspecifiedError, + ReasonString = pubCompPacket.Properties.ReasonString + }; + + result.UserProperties.AddRange(pubCompPacket.Properties.UserProperties); + return result; + } + + result = new MqttClientPublishResult + { + PacketIdentifier = pubCompPacket.PacketIdentifier, + ReasonCode = MqttClientPublishReasonCode.Success, + ReasonString = pubCompPacket.Properties.ReasonString + }; + + result.UserProperties.AddRange(pubCompPacket.Properties.UserProperties); + + if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success) + { + // Both enums share the same values. + result.ReasonCode = (MqttClientPublishReasonCode) pubRecPacket.ReasonCode; + } + + return result; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs index 2e229987..c6337a21 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs @@ -3,7 +3,7 @@ using MQTTnet.Packets; namespace MQTTnet.Client.Subscribing { - public class MqttClientSubscribeOptions + public sealed class MqttClientSubscribeOptions { /// /// Gets or sets a list of topic filters the client wants to subscribe to. diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs index 66d68d4e..b6141f39 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs @@ -1,9 +1,22 @@ using System.Collections.Generic; +using MQTTnet.Packets; namespace MQTTnet.Client.Subscribing { - public class MqttClientSubscribeResult + public sealed class MqttClientSubscribeResult { public List Items { get; } = new List(); + + /// + /// Gets the user properties which were part of the SUBACK packet. + /// MQTTv5 only. + /// + public List UserProperties { get; } = new List(); + + /// + /// Gets the reason string. + /// MQTTv5 only. + /// + public string ReasonString { get; internal set; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultCode.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultCode.cs index 119ec4fa..f1644b4e 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultCode.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultCode.cs @@ -2,10 +2,11 @@ { public enum MqttClientSubscribeResultCode { - GrantedQoS0 = 0, - GrantedQoS1 = 1, - GrantedQoS2 = 2, - UnspecifiedError = 128, + GrantedQoS0 = 0x00, + GrantedQoS1 = 0x01, + GrantedQoS2 = 0x02, + UnspecifiedError = 0x80, + ImplementationSpecificError = 131, NotAuthorized = 135, TopicFilterInvalid = 143, diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs new file mode 100644 index 00000000..cec62e65 --- /dev/null +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs @@ -0,0 +1,66 @@ +using System; +using System.Linq; +using MQTTnet.Exceptions; +using MQTTnet.Packets; + +namespace MQTTnet.Client.Subscribing +{ + public sealed class MqttClientSubscribeResultFactory + { + public MqttClientSubscribeResult Create(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) + { + if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); + if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket)); + + // MQTTv3.1.1 handling. + if (subAckPacket.ReturnCodes.Any() && subAckPacket.ReturnCodes.Count != subscribePacket.TopicFilters.Count) + { + throw new MqttProtocolViolationException( + "The return codes are not matching the topic filters [MQTT-3.9.3-1]."); + } + + // MQTTv5.0.0 handling. + if (subAckPacket.ReasonCodes.Any() && subAckPacket.ReasonCodes.Count != subscribePacket.TopicFilters.Count) + { + throw new MqttProtocolViolationException( + "The reason codes are not matching the topic filters [MQTT-3.9.3-1]."); + } + + var result = new MqttClientSubscribeResult + { + ReasonString = subAckPacket.Properties.ReasonString + }; + + result.UserProperties.AddRange(subAckPacket.Properties.UserProperties); + + for (var i = 0; i < subscribePacket.TopicFilters.Count; i++) + { + result.Items.Add(CreateSubscribeResultItem(i, subscribePacket, subAckPacket)); + } + + return result; + } + + static MqttClientSubscribeResultItem CreateSubscribeResultItem(int index, MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) + { + MqttClientSubscribeResultCode resultCode; + + if (subAckPacket.ReturnCodes.Any()) + { + // MQTTv3.1.1 handling. + resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReturnCodes[index]; + } + else + { + // MQTTv5.0.0 handling. + resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[index]; + } + + return new MqttClientSubscribeResultItem + { + TopicFilter = subscribePacket.TopicFilters[index], + ResultCode = resultCode + }; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs index 3e2d27f1..277bfe4c 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs @@ -1,25 +1,17 @@ -using System; - -namespace MQTTnet.Client.Subscribing +namespace MQTTnet.Client.Subscribing { - public class MqttClientSubscribeResultItem + public sealed class MqttClientSubscribeResultItem { - public MqttClientSubscribeResultItem(MqttTopicFilter topicFilter, MqttClientSubscribeResultCode resultCode) - { - TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); - ResultCode = resultCode; - } - /// /// Gets or sets the topic filter. /// The topic filter can contain topics and wildcards. /// - public MqttTopicFilter TopicFilter { get; } + public MqttTopicFilter TopicFilter { get; internal set; } /// /// Gets or sets the result code. /// Hint: MQTT 5 feature only. /// - public MqttClientSubscribeResultCode ResultCode { get; } + public MqttClientSubscribeResultCode ResultCode { get; internal set; } } } diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptions.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptions.cs index 09a4ce29..bf713701 100644 --- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptions.cs +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptions.cs @@ -3,7 +3,7 @@ using MQTTnet.Packets; namespace MQTTnet.Client.Unsubscribing { - public class MqttClientUnsubscribeOptions + public sealed class MqttClientUnsubscribeOptions { /// /// Gets or sets a list of topic filters the client wants to unsubscribe from. diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs index f3e4c4c9..d4ce4f6f 100644 --- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs @@ -1,9 +1,22 @@ using System.Collections.Generic; +using MQTTnet.Packets; namespace MQTTnet.Client.Unsubscribing { - public class MqttClientUnsubscribeResult + public sealed class MqttClientUnsubscribeResult { public List Items { get; } =new List(); + + /// + /// Gets the user properties which were part of the UNSUBACK packet. + /// MQTTv5 only. + /// + public List UserProperties { get; } = new List(); + + /// + /// Gets the reason string. + /// MQTTv5 only. + /// + public string ReasonString { get; internal set; } } } diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs new file mode 100644 index 00000000..a030bbdf --- /dev/null +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs @@ -0,0 +1,54 @@ +using System; +using System.Linq; +using MQTTnet.Exceptions; +using MQTTnet.Packets; + +namespace MQTTnet.Client.Unsubscribing +{ + public sealed class MqttClientUnsubscribeResultFactory + { + public MqttClientUnsubscribeResult Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket) + { + if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); + if (unsubAckPacket == null) throw new ArgumentNullException(nameof(unsubAckPacket)); + + // MQTTv3.1.1 has no reason code at all! + if (unsubAckPacket.ReasonCodes.Count != 0 && unsubAckPacket.ReasonCodes.Count != unsubscribePacket.TopicFilters.Count) + { + throw new MqttProtocolViolationException( + "The return codes are not matching the topic filters [MQTT-3.9.3-1]."); + } + + var result = new MqttClientUnsubscribeResult + { + ReasonString = unsubAckPacket.Properties.ReasonString + }; + + result.UserProperties.AddRange(unsubAckPacket.Properties.UserProperties); + + for (var i = 0; i < unsubscribePacket.TopicFilters.Count; i++) + { + result.Items.Add(CreateUnsubscribeResultItem(i, unsubscribePacket, unsubAckPacket)); + } + + return result; + } + + static MqttClientUnsubscribeResultItem CreateUnsubscribeResultItem(int index, MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket) + { + var resultCode = MqttClientUnsubscribeResultCode.Success; + + if (unsubAckPacket.ReasonCodes.Any()) + { + // MQTTv3.1.1 has no reason code and no return code!. + resultCode = (MqttClientUnsubscribeResultCode) unsubAckPacket.ReasonCodes[index]; + } + + return new MqttClientUnsubscribeResultItem + { + TopicFilter = unsubscribePacket.TopicFilters[index], + ResultCode = resultCode + }; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs index 065fb7f2..1cef26e0 100644 --- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs +++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs @@ -2,24 +2,21 @@ namespace MQTTnet.Client.Unsubscribing { - public class MqttClientUnsubscribeResultItem + public sealed class MqttClientUnsubscribeResultItem { - public MqttClientUnsubscribeResultItem(string topicFilter, MqttClientUnsubscribeResultCode reasonCode) - { - TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); - ReasonCode = reasonCode; - } - /// /// Gets or sets the topic filter. /// The topic filter can contain topics and wildcards. /// - public string TopicFilter { get; } + public string TopicFilter { get; internal set; } + + [Obsolete("Use ResultCode instead. This property will be removed soon.")] + public MqttClientUnsubscribeResultCode ReasonCode => ResultCode; /// /// Gets or sets the result code. /// Hint: MQTT 5 feature only. /// - public MqttClientUnsubscribeResultCode ReasonCode { get; } + public MqttClientUnsubscribeResultCode ResultCode { get; internal set; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/IMqttDataConverter.cs b/Source/MQTTnet/Formatter/IMqttDataConverter.cs deleted file mode 100644 index 4fea2562..00000000 --- a/Source/MQTTnet/Formatter/IMqttDataConverter.cs +++ /dev/null @@ -1,30 +0,0 @@ -using MQTTnet.Client.Connecting; -using MQTTnet.Client.Publishing; -using MQTTnet.Client.Subscribing; -using MQTTnet.Client.Unsubscribing; -using MQTTnet.Packets; -using MQTTnet.Server.Internal; - -namespace MQTTnet.Formatter -{ - public interface IMqttDataConverter - { - MqttClientConnectResult CreateClientConnectResult(MqttConnAckPacket connAckPacket); - - MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket); - - MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket); - - MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket); - - MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket); - - MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode); - - MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode); - - MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult); - - MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult); - } -} diff --git a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs index 6dd417c5..b4a26b84 100644 --- a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs +++ b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs @@ -6,8 +6,6 @@ namespace MQTTnet.Formatter { public interface IMqttPacketFormatter { - IMqttDataConverter DataConverter { get; } - ArraySegment Encode(MqttBasePacket mqttPacket); MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket); diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index 1e2df944..bd7d3820 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs @@ -29,17 +29,7 @@ namespace MQTTnet.Formatter } public MqttProtocolVersion ProtocolVersion { get; private set; } = MqttProtocolVersion.Unknown; - - public IMqttDataConverter DataConverter - { - get - { - ThrowIfFormatterNotSet(); - - return _formatter.DataConverter; - } - } - + public IMqttPacketWriter Writer { get; } public ArraySegment Encode(MqttBasePacket packet) diff --git a/Source/MQTTnet/Formatter/MqttPubCompPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubCompPacketFactory.cs new file mode 100644 index 00000000..03145e71 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttPubCompPacketFactory.cs @@ -0,0 +1,20 @@ +using System; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter +{ + public sealed class MqttPubCompPacketFactory + { + public MqttPubCompPacket Create(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + { + if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket)); + + return new MqttPubCompPacket + { + PacketIdentifier = pubRelPacket.PacketIdentifier, + ReasonCode = (MqttPubCompReasonCode) (int) reasonCode + }; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs new file mode 100644 index 00000000..08c4ca51 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs @@ -0,0 +1,20 @@ +using System; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter +{ + public sealed class MqttPubRelPacketFactory + { + public MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + { + if (pubRecPacket == null) throw new ArgumentNullException(nameof(pubRecPacket)); + + return new MqttPubRelPacket + { + PacketIdentifier = pubRecPacket.PacketIdentifier, + ReasonCode = (MqttPubRelReasonCode) (int) reasonCode + }; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs new file mode 100644 index 00000000..8396ba9b --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs @@ -0,0 +1,28 @@ +using System; +using MQTTnet.Packets; +using MQTTnet.Server.Internal; + +namespace MQTTnet.Formatter +{ + public sealed class MqttSubAckPacketFactory + { + public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult) + { + if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); + if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult)); + + var subAckPacket = new MqttSubAckPacket + { + PacketIdentifier = subscribePacket.PacketIdentifier + }; + + // MQTTv3.1.1 + subAckPacket.ReturnCodes.AddRange(subscribeResult.ReturnCodes); + + // MQTTv5.0.0 + subAckPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes); + + return subAckPacket; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs new file mode 100644 index 00000000..e64ca0a3 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs @@ -0,0 +1,25 @@ +using System; +using MQTTnet.Packets; +using MQTTnet.Server.Internal; + +namespace MQTTnet.Formatter +{ + public sealed class MqttUnsubAckPacketFactory + { + public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult) + { + if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); + if (unsubscribeResult == null) throw new ArgumentNullException(nameof(unsubscribeResult)); + + var unsubAckPacket = new MqttUnsubAckPacket + { + PacketIdentifier = unsubscribePacket.PacketIdentifier + }; + + // MQTTv5.0.0 only. + unsubAckPacket.ReasonCodes.AddRange(unsubscribeResult.ReasonCodes); + + return unsubAckPacket; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs deleted file mode 100644 index 6af35a6e..00000000 --- a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs +++ /dev/null @@ -1,176 +0,0 @@ -using MQTTnet.Client.Connecting; -using MQTTnet.Client.Publishing; -using MQTTnet.Client.Subscribing; -using MQTTnet.Client.Unsubscribing; -using MQTTnet.Exceptions; -using MQTTnet.Packets; -using MQTTnet.Protocol; -using MQTTnet.Server; -using System; -using System.Linq; -using MQTTnet.Server.Internal; - -namespace MQTTnet.Formatter.V3 -{ - public class MqttV310DataConverter : IMqttDataConverter - { - public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket)); - - return new MqttPubCompPacket - { - PacketIdentifier = pubRelPacket.PacketIdentifier - }; - } - - public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (pubRecPacket == null) throw new ArgumentNullException(nameof(pubRecPacket)); - - return new MqttPubRelPacket - { - PacketIdentifier = pubRecPacket.PacketIdentifier - }; - } - - public MqttClientConnectResult CreateClientConnectResult(MqttConnAckPacket connAckPacket) - { - if (connAckPacket == null) throw new ArgumentNullException(nameof(connAckPacket)); - - MqttClientConnectResultCode resultCode; - switch (connAckPacket.ReturnCode) - { - case MqttConnectReturnCode.ConnectionAccepted: - { - resultCode = MqttClientConnectResultCode.Success; - break; - } - - case MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion: - { - resultCode = MqttClientConnectResultCode.UnsupportedProtocolVersion; - break; - } - - case MqttConnectReturnCode.ConnectionRefusedNotAuthorized: - { - resultCode = MqttClientConnectResultCode.NotAuthorized; - break; - } - - case MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword: - { - resultCode = MqttClientConnectResultCode.BadUserNameOrPassword; - break; - } - - case MqttConnectReturnCode.ConnectionRefusedIdentifierRejected: - { - resultCode = MqttClientConnectResultCode.ClientIdentifierNotValid; - break; - } - - case MqttConnectReturnCode.ConnectionRefusedServerUnavailable: - { - resultCode = MqttClientConnectResultCode.ServerUnavailable; - break; - } - - default: - throw new MqttProtocolViolationException("Received unexpected return code."); - } - - return new MqttClientConnectResult - { - RetainAvailable = true, // Always true because v3.1.1 does not have a way to "disable" that feature. - WildcardSubscriptionAvailable = true, // Always true because v3.1.1 does not have a way to "disable" that feature. - IsSessionPresent = connAckPacket.IsSessionPresent, - ResultCode = resultCode - }; - } - - public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) - { - if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); - if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket)); - - if (subAckPacket.ReturnCodes.Count != subscribePacket.TopicFilters.Count) - { - throw new MqttProtocolViolationException("The return codes are not matching the topic filters [MQTT-3.9.3-1]."); - } - - var result = new MqttClientSubscribeResult(); - - result.Items.AddRange(subscribePacket.TopicFilters.Select((t, i) => - new MqttClientSubscribeResultItem(t, (MqttClientSubscribeResultCode)subAckPacket.ReturnCodes[i]))); - - return result; - } - - public MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket) - { - if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - if (unsubAckPacket == null) throw new ArgumentNullException(nameof(unsubAckPacket)); - - var result = new MqttClientUnsubscribeResult(); - - result.Items.AddRange(unsubscribePacket.TopicFilters.Select((t, i) => - new MqttClientUnsubscribeResultItem(t, MqttClientUnsubscribeResultCode.Success))); - - return result; - } - - 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 - { - PacketIdentifier = subscribePacket.PacketIdentifier - }; - - subAckPacket.ReturnCodes.AddRange(subscribeResult.ReturnCodes); - - return subAckPacket; - } - - public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult _) - { - if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - - // The UnsubscribeResult is not supported in protocol version 3.1.1! - return new MqttUnsubAckPacket - { - PacketIdentifier = unsubscribePacket.PacketIdentifier - }; - } - - public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket) - { - return new MqttClientPublishResult - { - PacketIdentifier = pubAckPacket?.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.Success - }; - } - - public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) - { - if (pubRecPacket == null || pubCompPacket == null) - { - return new MqttClientPublishResult - { - ReasonCode = MqttClientPublishReasonCode.UnspecifiedError - }; - } - - return new MqttClientPublishResult - { - PacketIdentifier = pubCompPacket.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.Success - }; - } - } -} diff --git a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs index 16fb380e..42aeee20 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs @@ -1,6 +1,5 @@ using System; using System.Linq; -using System.Net; using System.Runtime.CompilerServices; using MQTTnet.Adapter; using MQTTnet.Exceptions; @@ -12,9 +11,7 @@ namespace MQTTnet.Formatter.V3 public class MqttV310PacketFormatter : IMqttPacketFormatter { const int FixedHeaderSize = 1; - - static readonly MqttPingReqPacket PingReqPacket = new MqttPingReqPacket(); - static readonly MqttPingRespPacket PingRespPacket = new MqttPingRespPacket(); + static readonly MqttDisconnectPacket DisconnectPacket = new MqttDisconnectPacket(); readonly IMqttPacketWriter _packetWriter; @@ -24,8 +21,6 @@ namespace MQTTnet.Formatter.V3 _packetWriter = packetWriter; } - public IMqttDataConverter DataConverter { get; } = new MqttV310DataConverter(); - public ArraySegment Encode(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -72,8 +67,8 @@ namespace MQTTnet.Formatter.V3 case MqttControlPacketType.PubRec: return DecodePubRecPacket(receivedMqttPacket.BodyReader); case MqttControlPacketType.PubRel: return DecodePubRelPacket(receivedMqttPacket.BodyReader); case MqttControlPacketType.PubComp: return DecodePubCompPacket(receivedMqttPacket.BodyReader); - case MqttControlPacketType.PingReq: return PingReqPacket; - case MqttControlPacketType.PingResp: return PingRespPacket; + case MqttControlPacketType.PingReq: return MqttPingReqPacket.Instance; + case MqttControlPacketType.PingResp: return MqttPingRespPacket.Instance; case MqttControlPacketType.Subscribe: return DecodeSubscribePacket(receivedMqttPacket.BodyReader); case MqttControlPacketType.SubAck: return DecodeSubAckPacket(receivedMqttPacket.BodyReader); case MqttControlPacketType.Unsubscibe: return DecodeUnsubscribePacket(receivedMqttPacket.BodyReader); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs deleted file mode 100644 index 79e97168..00000000 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ /dev/null @@ -1,202 +0,0 @@ -using MQTTnet.Client.Connecting; -using MQTTnet.Client.Publishing; -using MQTTnet.Client.Subscribing; -using MQTTnet.Client.Unsubscribing; -using MQTTnet.Exceptions; -using MQTTnet.Packets; -using MQTTnet.Protocol; -using System; -using System.Collections.Generic; -using System.Linq; -using MQTTnet.Server.Internal; - -namespace MQTTnet.Formatter.V5 -{ - public sealed class MqttV500DataConverter : IMqttDataConverter - { - public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, - MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket)); - - return new MqttPubCompPacket - { - PacketIdentifier = pubRelPacket.PacketIdentifier, - ReasonCode = (MqttPubCompReasonCode) (int) reasonCode - }; - } - - public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, - MqttApplicationMessageReceivedReasonCode reasonCode) - { - if (pubRecPacket == null) throw new ArgumentNullException(nameof(pubRecPacket)); - - return new MqttPubRelPacket - { - PacketIdentifier = pubRecPacket.PacketIdentifier, - ReasonCode = (MqttPubRelReasonCode) (int) reasonCode - }; - } - - public MqttClientConnectResult CreateClientConnectResult(MqttConnAckPacket connAckPacket) - { - if (connAckPacket == null) throw new ArgumentNullException(nameof(connAckPacket)); - - return new MqttClientConnectResult - { - IsSessionPresent = connAckPacket.IsSessionPresent, - ResultCode = (MqttClientConnectResultCode) (int) connAckPacket.ReasonCode, - WildcardSubscriptionAvailable = connAckPacket.Properties?.WildcardSubscriptionAvailable ?? true, - RetainAvailable = connAckPacket.Properties?.RetainAvailable ?? true, - AssignedClientIdentifier = connAckPacket.Properties?.AssignedClientIdentifier, - AuthenticationMethod = connAckPacket.Properties?.AuthenticationMethod, - AuthenticationData = connAckPacket.Properties?.AuthenticationData, - MaximumPacketSize = connAckPacket.Properties?.MaximumPacketSize, - ReasonString = connAckPacket.Properties?.ReasonString, - ReceiveMaximum = connAckPacket.Properties?.ReceiveMaximum, - MaximumQoS = connAckPacket.Properties?.MaximumQoS ?? MqttQualityOfServiceLevel.ExactlyOnce, - ResponseInformation = connAckPacket.Properties?.ResponseInformation, - TopicAliasMaximum = connAckPacket.Properties?.TopicAliasMaximum ?? 0, - ServerReference = connAckPacket.Properties?.ServerReference, - ServerKeepAlive = connAckPacket.Properties?.ServerKeepAlive, - SessionExpiryInterval = connAckPacket.Properties?.SessionExpiryInterval, - SubscriptionIdentifiersAvailable = connAckPacket.Properties?.SubscriptionIdentifiersAvailable ?? true, - SharedSubscriptionAvailable = connAckPacket.Properties?.SharedSubscriptionAvailable ?? true, - UserProperties = connAckPacket.Properties?.UserProperties ?? new List() - }; - } - - public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, - MqttSubAckPacket subAckPacket) - { - if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); - if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket)); - - if (subAckPacket.ReasonCodes.Count != subscribePacket.TopicFilters.Count) - { - throw new MqttProtocolViolationException( - "The reason codes are not matching the topic filters [MQTT-3.9.3-1]."); - } - - var result = new MqttClientSubscribeResult(); - - result.Items.AddRange(subscribePacket.TopicFilters.Select((t, i) => - new MqttClientSubscribeResultItem(t, (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[i]))); - - return result; - } - - public MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, - MqttUnsubAckPacket unsubAckPacket) - { - if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - if (unsubAckPacket == null) throw new ArgumentNullException(nameof(unsubAckPacket)); - - if (unsubAckPacket.ReasonCodes.Count != unsubscribePacket.TopicFilters.Count) - { - throw new MqttProtocolViolationException( - "The return codes are not matching the topic filters [MQTT-3.9.3-1]."); - } - - var result = new MqttClientUnsubscribeResult(); - - result.Items.AddRange(unsubscribePacket.TopicFilters.Select((t, i) => - new MqttClientUnsubscribeResultItem(t, - (MqttClientUnsubscribeResultCode) unsubAckPacket.ReasonCodes[i]))); - - return result; - } - - 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 - { - PacketIdentifier = subscribePacket.PacketIdentifier - }; - - subAckPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes); - - return subAckPacket; - } - - public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, - UnsubscribeResult unsubscribeResult) - { - if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - if (unsubscribeResult == null) throw new ArgumentNullException(nameof(unsubscribeResult)); - - var unsubAckPacket = new MqttUnsubAckPacket - { - PacketIdentifier = unsubscribePacket.PacketIdentifier - }; - - unsubAckPacket.ReasonCodes.AddRange(unsubscribeResult.ReasonCodes); - - return unsubAckPacket; - } - - public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket) - { - var result = new MqttClientPublishResult - { - ReasonCode = MqttClientPublishReasonCode.Success, - ReasonString = pubAckPacket?.Properties?.ReasonString, - UserProperties = pubAckPacket?.Properties?.UserProperties - }; - - if (pubAckPacket != null) - { - // 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; - - result.PacketIdentifier = pubAckPacket.PacketIdentifier; - } - - return result; - } - - public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, - MqttPubCompPacket pubCompPacket) - { - if (pubRecPacket == null || pubCompPacket == null) - { - return new MqttClientPublishResult - { - ReasonCode = MqttClientPublishReasonCode.UnspecifiedError - }; - } - - // The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC. - if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound) - { - return new MqttClientPublishResult - { - PacketIdentifier = pubCompPacket.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.UnspecifiedError, - ReasonString = pubCompPacket.Properties?.ReasonString, - UserProperties = pubCompPacket.Properties?.UserProperties - }; - } - - var result = new MqttClientPublishResult - { - PacketIdentifier = pubCompPacket.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.Success, - ReasonString = pubCompPacket.Properties?.ReasonString, - UserProperties = pubCompPacket.Properties?.UserProperties - }; - - if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success) - { - // Both enums share the same values. - result.ReasonCode = (MqttClientPublishReasonCode) pubRecPacket.ReasonCode; - } - - return result; - } - } -} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs index a34aa536..f5454963 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs @@ -13,8 +13,6 @@ namespace MQTTnet.Formatter.V5 { _encoder = new MqttV500PacketEncoder(writer); } - - public IMqttDataConverter DataConverter { get; } = new MqttV500DataConverter(); public ArraySegment Encode(MqttBasePacket mqttPacket) { diff --git a/Source/MQTTnet/Protocol/MqttSubscribeReasonCode.cs b/Source/MQTTnet/Protocol/MqttSubscribeReasonCode.cs index e3f2c0f9..e0b9407c 100644 --- a/Source/MQTTnet/Protocol/MqttSubscribeReasonCode.cs +++ b/Source/MQTTnet/Protocol/MqttSubscribeReasonCode.cs @@ -2,10 +2,13 @@ { public enum MqttSubscribeReasonCode { - GrantedQoS0 = 0, - GrantedQoS1 = 1, - GrantedQoS2 = 2, - UnspecifiedError = 128, + // Compatible with MQTTv3.1.1. + GrantedQoS0 = 0x00, + GrantedQoS1 = 0x01, + GrantedQoS2 = 0x02, + UnspecifiedError = 0x80, + + // New in MQTTv5. ImplementationSpecificError = 131, NotAuthorized = 135, TopicFilterInvalid = 143, diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 6541891d..5ea2cf52 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -3,12 +3,10 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; -using MQTTnet.Client; 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; @@ -20,7 +18,6 @@ namespace MQTTnet.Server.Internal public sealed class MqttClientConnection : IDisposable { readonly Dictionary _topicAlias = new Dictionary(); - readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly MqttClientSessionsManager _sessionsManager; @@ -162,7 +159,7 @@ namespace MQTTnet.Server.Internal if (packet is MqttPublishPacket publishPacket) { - await HandleIncomingPublishPacket(publishPacket, cancellationToken).ConfigureAwait(false); + HandleIncomingPublishPacket(publishPacket); } else if (packet is MqttPubAckPacket pubAckPacket) { @@ -176,7 +173,9 @@ namespace MQTTnet.Server.Internal { // TODO: Do anything! // TODO: Send pub rel! - Session.EnqueuePacket(new MqttPacketBusItem(ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success))); + var pubRelPacketFactory = new MqttPubRelPacketFactory(); + var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + Session.EnqueuePacket(new MqttPacketBusItem(pubRelPacket)); } else if (packet is MqttPubRelPacket pubRelPacket) { @@ -390,16 +389,21 @@ namespace MQTTnet.Server.Internal Task HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { - var pubCompPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); - return SendPacketAsync(pubCompPacket, cancellationToken); + var pubCompPacketFactory = new MqttPubCompPacketFactory(); + var pubCompPacket =pubCompPacketFactory.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); + Session.EnqueuePacket(new MqttPacketBusItem(pubCompPacket)); + + //return SendPacketAsync(pubCompPacket, cancellationToken); + return PlatformAbstractionLayer.CompletedTask; } async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { - var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket).ConfigureAwait(false); - var subAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateSubAckPacket(subscribePacket, subscribeResult); + var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false); + + var subAckPacketFactory = new MqttSubAckPacketFactory(); + var subAckPacket = subAckPacketFactory.Create(subscribePacket, subscribeResult); - //await SendPacketAsync(subAckPacket, cancellationToken).ConfigureAwait(false); Session.EnqueuePacket(new MqttPacketBusItem(subAckPacket)); if (subscribeResult.CloseConnection) @@ -411,26 +415,27 @@ namespace MQTTnet.Server.Internal var publishPacketFactory = new MqttPublishPacketFactory(); foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages) { - //var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(retainedApplicationMessage.ApplicationMessage); - Session.EnqueuePacket(new MqttPacketBusItem(publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage))); + var publishPacket = publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage); + Session.EnqueuePacket(new MqttPacketBusItem(publishPacket)); } } async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { - var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket).ConfigureAwait(false); - var unsubAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, unsubscribeResult); + var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); + + var unsubAckPacketFactory = new MqttUnsubAckPacketFactory(); + var unsubAckPacket = unsubAckPacketFactory.Create(unsubscribePacket, unsubscribeResult); Session.EnqueuePacket(new MqttPacketBusItem(unsubAckPacket)); - //await SendPacketAsync(unsubAckPacket, cancellationToken).ConfigureAwait(false); - + if (unsubscribeResult.CloseConnection) { StopInternal(); } } - Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + void HandleIncomingPublishPacket(MqttPublishPacket publishPacket) { HandleTopicAlias(publishPacket); @@ -452,17 +457,13 @@ namespace MQTTnet.Server.Internal var pubAckPacketFactory = new MqttPubAckPacketFactory(); var pubAckPacket = pubAckPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket)); - //return SendPacketAsync(pubAckPacket, cancellationToken); break; } case MqttQualityOfServiceLevel.ExactlyOnce: { - // TODO: Migrate to packet bus. var pubRecPacketFactory = new MqttPubRecPacketFactory(); var pubRecPacket = pubRecPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket)); - - //return SendPacketAsync(pubRecPacket, cancellationToken); break; } default: @@ -470,8 +471,6 @@ namespace MQTTnet.Server.Internal throw new MqttCommunicationException("Received a not supported QoS level."); } } - - return PlatformAbstractionLayer.CompletedTask; } void HandleTopicAlias(MqttPublishPacket publishPacket) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index e6bb2d0b..f914904c 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -236,7 +236,7 @@ namespace MQTTnet.Server.Internal var clientSession = GetClientSession(clientId); - var subscribeResult = await clientSession.SubscriptionsManager.Subscribe(fakeSubscribePacket).ConfigureAwait(false); + var subscribeResult = await clientSession.SubscriptionsManager.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false); foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages) { @@ -257,7 +257,7 @@ namespace MQTTnet.Server.Internal var fakeUnsubscribePacket = new MqttUnsubscribePacket(); fakeUnsubscribePacket.TopicFilters.AddRange(topicFilters); - return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket); + return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket, CancellationToken.None); } public async Task DeleteSessionAsync(string clientId) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index abf8d725..5c895cdc 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Server.Internal _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); } - public async Task Subscribe(MqttSubscribePacket subscribePacket) + public async Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); @@ -45,7 +45,7 @@ namespace MQTTnet.Server.Internal // lower one. foreach (var originalTopicFilter in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel)) { - var interceptorContext = await InterceptSubscribe(originalTopicFilter).ConfigureAwait(false); + var interceptorContext = await InterceptSubscribe(originalTopicFilter, cancellationToken).ConfigureAwait(false); var finalTopicFilter = interceptorContext.TopicFilter; var acceptSubscription = interceptorContext.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; @@ -91,7 +91,7 @@ namespace MQTTnet.Server.Internal return result; } - public async Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket) + public async Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); @@ -104,7 +104,7 @@ namespace MQTTnet.Server.Internal { _subscriptions.TryGetValue(topicFilter, out var existingSubscription); - var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription).ConfigureAwait(false); + var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription, cancellationToken).ConfigureAwait(false); var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success; result.ReasonCodes.Add(interceptorContext.ReasonCode); @@ -292,14 +292,15 @@ namespace MQTTnet.Server.Internal } } - async Task InterceptSubscribe(MqttTopicFilter topicFilter) + async Task InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken) { var context = new MqttSubscriptionInterceptorContext { ClientId = _clientSession.ClientId, TopicFilter = topicFilter, SessionItems = _clientSession.Items, - Session = new MqttSessionStatus(_clientSession) + Session = new MqttSessionStatus(_clientSession), + CancellationToken = cancellationToken }; if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) @@ -338,13 +339,14 @@ namespace MQTTnet.Server.Internal return context; } - async Task InterceptUnsubscribe(string topicFilter, Subscription subscription) + async Task InterceptUnsubscribe(string topicFilter, Subscription subscription, CancellationToken cancellationToken) { var context = new MqttUnsubscriptionInterceptorContext { ClientId = _clientSession.ClientId, Topic = topicFilter, - SessionItems = _clientSession.Items + SessionItems = _clientSession.Items, + CancellationToken = cancellationToken }; if (subscription == null) diff --git a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs index 3ce23b1a..b7098001 100644 --- a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using MQTTnet.Protocol; using MQTTnet.Server.Status; @@ -74,5 +75,10 @@ namespace MQTTnet.Server /// Gets the current client session. /// public IMqttSessionStatus Session { get; internal set; } + + /// + /// Gets the cancellation token which can indicate that the client connection gets down. + /// + public CancellationToken CancellationToken { get; internal set; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs index 9b9cb104..19ac121a 100644 --- a/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using MQTTnet.Protocol; namespace MQTTnet.Server @@ -51,5 +52,10 @@ namespace MQTTnet.Server /// Gets or sets whether the broker should close the client connection. /// public bool CloseConnection { get; set; } + + /// + /// Gets the cancellation token which can indicate that the client connection gets down. + /// + public CancellationToken CancellationToken { get; internal set; } } } diff --git a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs index b2272770..4d930dc4 100644 --- a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Packets; @@ -22,7 +23,7 @@ namespace MQTTnet.Tests.Server var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build()); - await sm.Subscribe(sp); + await sm.Subscribe(sp, CancellationToken.None); var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, ""); Assert.IsTrue(result.IsSubscribed); @@ -39,7 +40,7 @@ namespace MQTTnet.Tests.Server var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); - await sm.Subscribe(sp); + await sm.Subscribe(sp, CancellationToken.None); var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, ""); Assert.IsTrue(result.IsSubscribed); @@ -57,7 +58,7 @@ namespace MQTTnet.Tests.Server sp.TopicFilters.Add(new MqttTopicFilter { Topic = "#", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); sp.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); - await sm.Subscribe(sp); + await sm.Subscribe(sp, CancellationToken.None); var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, ""); Assert.IsTrue(result.IsSubscribed); @@ -74,7 +75,7 @@ namespace MQTTnet.Tests.Server var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build()); - await sm.Subscribe(sp); + await sm.Subscribe(sp, CancellationToken.None); Assert.IsFalse(sm.CheckSubscriptions("A/B/X", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); } @@ -89,13 +90,13 @@ namespace MQTTnet.Tests.Server var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build()); - await sm.Subscribe(sp); + await sm.Subscribe(sp, CancellationToken.None); Assert.IsTrue(sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); var up = new MqttUnsubscribePacket(); up.TopicFilters.Add("A/B/C"); - await sm.Unsubscribe(up); + await sm.Unsubscribe(up, CancellationToken.None); Assert.IsFalse(sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); }