diff --git a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectReason.cs b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectReason.cs index e1ea30d7..cce2377f 100644 --- a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectReason.cs +++ b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectReason.cs @@ -12,7 +12,7 @@ ServerBusy = 137, ServerShuttingDown = 139, BadAuthenticationMethod = 140, - KeepaliveTimeout = 141, + KeepAliveTimeout = 141, SessionTakenOver = 142, TopicFilterInvalid = 143, TopicNameInvalid = 144, diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 3414e990..b16cac7e 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -167,7 +167,7 @@ namespace MQTTnet.Client public Task PingAsync(CancellationToken cancellationToken) { - return SendAndReceiveAsync(new MqttPingReqPacket(), cancellationToken); + return SendAndReceiveAsync(MqttPingReqPacket.Instance, cancellationToken); } public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken) @@ -650,11 +650,8 @@ namespace MQTTnet.Client if (!eventArgs.ProcessingFailed) { - await SendAsync(new MqttPubAckPacket - { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = (MqttPubAckReasonCode)eventArgs.ReasonCode - }, cancellationToken).ConfigureAwait(false); + var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, eventArgs.ReasonCode); + await SendAsync(pubAckPacket, cancellationToken).ConfigureAwait(false); } } else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) @@ -663,11 +660,8 @@ namespace MQTTnet.Client if (!eventArgs.ProcessingFailed) { - await SendAsync(new MqttPubRecPacket - { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = (MqttPubRecReasonCode)eventArgs.ReasonCode - }, cancellationToken).ConfigureAwait(false); + var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, eventArgs.ReasonCode); + await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false); } } else @@ -691,11 +685,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. - return SendAsync(new MqttPubRelPacket - { - PacketIdentifier = pubRecPacket.PacketIdentifier, - ReasonCode = MqttPubRelReasonCode.PacketIdentifierNotFound - }, cancellationToken); + var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); + return SendAsync(pubRelPacket, cancellationToken); } return PlatformAbstractionLayer.CompletedTask; @@ -703,11 +694,8 @@ namespace MQTTnet.Client Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { - return SendAsync(new MqttPubCompPacket - { - PacketIdentifier = pubRelPacket.PacketIdentifier, - ReasonCode = MqttPubCompReasonCode.Success - }, cancellationToken); + var pubCompPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); + return SendAsync(pubCompPacket, cancellationToken); } Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) @@ -740,14 +728,14 @@ 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.CreatePublishResult(null); + return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(null); } async Task PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var response = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(response); + return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(response); } async Task PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -756,15 +744,11 @@ namespace MQTTnet.Client var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - var pubRelPacket = new MqttPubRelPacket - { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubRelReasonCode.Success - }; - + var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); - return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket); + return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubRecPacket, pubCompPacket); } async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) diff --git a/Source/MQTTnet/Formatter/IMqttDataConverter.cs b/Source/MQTTnet/Formatter/IMqttDataConverter.cs index 76079a1b..aec47027 100644 --- a/Source/MQTTnet/Formatter/IMqttDataConverter.cs +++ b/Source/MQTTnet/Formatter/IMqttDataConverter.cs @@ -14,24 +14,32 @@ namespace MQTTnet.Formatter { public interface IMqttDataConverter { - MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage); - - MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket); - - MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket); - MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket); MqttClientAuthenticateResult 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); + MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, IMqttClientOptions options); MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext); + + MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage); - MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket); - - MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket); + MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode); + + MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode); + MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode); + + MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode); + MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options); MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, Server.MqttClientSubscribeResult subscribeResult); @@ -41,9 +49,5 @@ namespace MQTTnet.Formatter MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List reasonCodes); MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options); - - MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket); - - MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket); } } diff --git a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs index 31ea9ed4..c2b16747 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs @@ -31,21 +31,43 @@ namespace MQTTnet.Formatter.V3 }; } - public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket) + public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + return new MqttPubAckPacket { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubAckReasonCode.Success + PacketIdentifier = publishPacket.PacketIdentifier }; } - public MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket) + public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + return new MqttPubRecPacket { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubRecReasonCode.Success + PacketIdentifier = publishPacket.PacketIdentifier + }; + } + + 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 }; } @@ -58,7 +80,8 @@ namespace MQTTnet.Formatter.V3 Topic = publishPacket.Topic, Payload = publishPacket.Payload, QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, - Retain = publishPacket.Retain + Retain = publishPacket.Retain, + Dup = publishPacket.Dup }; } @@ -184,6 +207,9 @@ namespace MQTTnet.Formatter.V3 public MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, Server.MqttClientSubscribeResult subscribeResult) { + if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); + if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult)); + var subackPacket = new MqttSubAckPacket { PacketIdentifier = subscribePacket.PacketIdentifier @@ -206,6 +232,9 @@ namespace MQTTnet.Formatter.V3 public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List reasonCodes) { + if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); + if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes)); + return new MqttUnsubAckPacket { PacketIdentifier = unsubscribePacket.PacketIdentifier, @@ -223,7 +252,7 @@ namespace MQTTnet.Formatter.V3 return new MqttDisconnectPacket(); } - public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket) + public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket) { return new MqttClientPublishResult { @@ -232,7 +261,7 @@ namespace MQTTnet.Formatter.V3 }; } - public MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) + public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) { if (pubRecPacket == null || pubCompPacket == null) { diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index 038a4ed9..49da6fcd 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -15,7 +15,7 @@ using MqttClientSubscribeResult = MQTTnet.Client.Subscribing.MqttClientSubscribe namespace MQTTnet.Formatter.V5 { - public class MqttV500DataConverter : IMqttDataConverter + public sealed class MqttV500DataConverter : IMqttDataConverter { public MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage) { @@ -27,7 +27,7 @@ namespace MQTTnet.Formatter.V5 Payload = applicationMessage.Payload, QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, Retain = applicationMessage.Retain, - Dup = false, + Dup = applicationMessage.Dup, Properties = new MqttPublishPacketProperties { ContentType = applicationMessage.ContentType, @@ -49,32 +49,61 @@ namespace MQTTnet.Formatter.V5 return packet; } - public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket) + public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + return new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubAckReasonCode.Success + ReasonCode = (MqttPubAckReasonCode)(int)reasonCode }; } - public MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket) + public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode) { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + return new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubRecReasonCode.Success + ReasonCode = (MqttPubRecReasonCode)(int)reasonCode + }; + } + + public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + { + if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket)); + + return new MqttPubCompPacket + { + PacketIdentifier = pubRelPacket.PacketIdentifier, + ReasonCode = (MqttPubCompReasonCode)(int)reasonCode + }; + } + + 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 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, @@ -93,7 +122,7 @@ namespace MQTTnet.Formatter.V5 return new MqttClientAuthenticateResult { IsSessionPresent = connAckPacket.IsSessionPresent, - ResultCode = (MqttClientConnectResultCode)connAckPacket.ReasonCode.Value, + ResultCode = (MqttClientConnectResultCode)(int)(connAckPacket.ReasonCode ?? 0), WildcardSubscriptionAvailable = connAckPacket.Properties?.WildcardSubscriptionAvailable, RetainAvailable = connAckPacket.Properties?.RetainAvailable, AssignedClientIdentifier = connAckPacket.Properties?.AssignedClientIdentifier, @@ -144,6 +173,8 @@ namespace MQTTnet.Formatter.V5 public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext) { + if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext)); + return new MqttConnAckPacket { ReasonCode = connectionValidatorContext.ReasonCode, @@ -213,6 +244,9 @@ namespace MQTTnet.Formatter.V5 public MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, Server.MqttClientSubscribeResult subscribeResult) { + if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); + if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult)); + var subackPacket = new MqttSubAckPacket { PacketIdentifier = subscribePacket.PacketIdentifier @@ -240,6 +274,9 @@ namespace MQTTnet.Formatter.V5 public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List reasonCodes) { + if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); + if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes)); + return new MqttUnsubAckPacket { PacketIdentifier = unsubscribePacket.PacketIdentifier, @@ -263,7 +300,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket) + public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket) { var result = new MqttClientPublishResult { @@ -276,7 +313,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)pubAckPacket.ReasonCode; + result.ReasonCode = (MqttClientPublishReasonCode)(int)(pubAckPacket.ReasonCode ?? 0); result.PacketIdentifier = pubAckPacket.PacketIdentifier; } @@ -284,7 +321,7 @@ namespace MQTTnet.Formatter.V5 return result; } - public MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) + public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) { if (pubRecPacket == null || pubCompPacket == null) { @@ -317,10 +354,10 @@ namespace MQTTnet.Formatter.V5 if (pubRecPacket.ReasonCode.HasValue) { // Both enums share the same values. - result.ReasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode.Value; + result.ReasonCode = (MqttClientPublishReasonCode)(pubRecPacket.ReasonCode ?? 0); } return result; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/MqttApplicationMessageReceivedReasonCode.cs b/Source/MQTTnet/MqttApplicationMessageReceivedReasonCode.cs index b192d311..3cd6e69d 100644 --- a/Source/MQTTnet/MqttApplicationMessageReceivedReasonCode.cs +++ b/Source/MQTTnet/MqttApplicationMessageReceivedReasonCode.cs @@ -9,6 +9,7 @@ NotAuthorized = 135, TopicNameInvalid = 144, PacketIdentifierInUse = 145, + PacketIdentifierNotFound = 146, QuotaExceeded = 151, PayloadFormatInvalid = 153 } diff --git a/Source/MQTTnet/Packets/MqttPingReqPacket.cs b/Source/MQTTnet/Packets/MqttPingReqPacket.cs index 459e8b91..18a3a162 100644 --- a/Source/MQTTnet/Packets/MqttPingReqPacket.cs +++ b/Source/MQTTnet/Packets/MqttPingReqPacket.cs @@ -3,7 +3,7 @@ public sealed class MqttPingReqPacket : MqttBasePacket { // This is a minor performance improvement. - public static MqttPingReqPacket Instance = new MqttPingReqPacket(); + public static readonly MqttPingReqPacket Instance = new MqttPingReqPacket(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttPingRespPacket.cs b/Source/MQTTnet/Packets/MqttPingRespPacket.cs index de1db5e2..8ddc349f 100644 --- a/Source/MQTTnet/Packets/MqttPingRespPacket.cs +++ b/Source/MQTTnet/Packets/MqttPingRespPacket.cs @@ -3,7 +3,7 @@ public sealed class MqttPingRespPacket : MqttBasePacket { // This is a minor performance improvement. - public static MqttPingRespPacket Instance = new MqttPingRespPacket(); + public static readonly MqttPingRespPacket Instance = new MqttPingRespPacket(); public override string ToString() { diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index bd31f4b7..75083a54 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -13,6 +13,7 @@ using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Client.Disconnecting; namespace MQTTnet.Server { @@ -42,7 +43,7 @@ namespace MQTTnet.Server long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere. long _receivedApplicationMessagesCount; long _sentApplicationMessagesCount; - MqttDisconnectReasonCode _disconnectReason; + MqttClientDisconnectReason _disconnectReason; public MqttClientConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, @@ -84,12 +85,12 @@ namespace MQTTnet.Server public MqttClientSession Session { get; } - public async Task StopAsync(MqttDisconnectReasonCode reason) + public async Task StopAsync(MqttClientDisconnectReason reason) { Status = MqttClientConnectionStatus.Finalizing; _disconnectReason = reason; - if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout) + if (reason == MqttClientDisconnectReason.SessionTakenOver || reason == MqttClientDisconnectReason.KeepAliveTimeout) { // Is is very important to send the DISCONNECT packet here BEFORE cancelling the // token because the entire connection is closed (disposed) as soon as the cancellation @@ -97,10 +98,18 @@ namespace MQTTnet.Server // at the client! try { - await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket + var disconnectOptions = new MqttClientDisconnectOptions { - ReasonCode = reason - }, CancellationToken.None).ConfigureAwait(false); + ReasonCode = reason, + ReasonString = reason.ToString() + }; + + var disconnectPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(disconnectOptions); + + using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout)) + { + await _channelAdapter.SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false); + } } catch (Exception exception) { @@ -235,7 +244,7 @@ namespace MQTTnet.Server } finally { - if (_disconnectReason == MqttDisconnectReasonCode.SessionTakenOver) + if (_disconnectReason == MqttClientDisconnectReason.SessionTakenOver) { disconnectType = MqttClientDisconnectType.Takeover; } @@ -277,12 +286,7 @@ namespace MQTTnet.Server Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { - var pubCompPacket = new MqttPubCompPacket - { - PacketIdentifier = pubRelPacket.PacketIdentifier, - ReasonCode = MqttPubCompReasonCode.Success - }; - + var pubCompPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); return SendAsync(pubCompPacket, cancellationToken); } @@ -327,12 +331,12 @@ namespace MQTTnet.Server } case MqttQualityOfServiceLevel.AtLeastOnce: { - var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket); + var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); return SendAsync(pubAckPacket, cancellationToken); } case MqttQualityOfServiceLevel.ExactlyOnce: { - var pubRecPacket = _dataConverter.CreatePubRecPacket(publishPacket); + var pubRecPacket = _dataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); return SendAsync(pubRecPacket, cancellationToken); } default: @@ -441,13 +445,10 @@ namespace MQTTnet.Server using (var awaiter2 = _packetDispatcher.AddAwaiter(publishPacket.PacketIdentifier)) { await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false); - await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); + var pubRecPacket = await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); - await SendAsync(new MqttPubRelPacket - { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubRelReasonCode.Success - }, cancellationToken).ConfigureAwait(false); + var pubRelPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + await SendAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 619f65fc..2c3202bb 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -12,6 +12,7 @@ using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Client.Disconnecting; using OperationCanceledException = System.OperationCanceledException; namespace MQTTnet.Server @@ -121,7 +122,7 @@ namespace MQTTnet.Server foreach (var connection in connections) { - await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false); + await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); } } @@ -211,7 +212,7 @@ namespace MQTTnet.Server if (connection != null) { - await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false); + await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); } _logger.Verbose("Session for client '{0}' deleted.", clientId); @@ -301,7 +302,7 @@ namespace MQTTnet.Server { if (clientConnection != null) { - await clientConnection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false); + await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); } } @@ -424,7 +425,7 @@ namespace MQTTnet.Server _connections[connectPacket.ClientId] = connection; } - existingConnection?.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).GetAwaiter().GetResult(); + existingConnection?.StopAsync(MqttClientDisconnectReason.SessionTakenOver).GetAwaiter().GetResult(); return connection; } diff --git a/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs index bd6eaa0a..7f967e9e 100644 --- a/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs @@ -3,8 +3,8 @@ using MQTTnet.Internal; using System; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Client.Disconnecting; using MQTTnet.Implementations; -using MQTTnet.Protocol; namespace MQTTnet.Server { @@ -108,7 +108,7 @@ namespace MQTTnet.Server // We do not need to wait for the task so no await is needed. // Also the internal state of the connection must be swapped to "Finalizing" because the // next iteration of the keep alive timer happens. - var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout); + var _ = connection.StopAsync(MqttClientDisconnectReason.KeepAliveTimeout); } catch (Exception exception) { diff --git a/Source/MQTTnet/Server/Status/MqttClientStatus.cs b/Source/MQTTnet/Server/Status/MqttClientStatus.cs index eca5d1a5..9f2f670e 100644 --- a/Source/MQTTnet/Server/Status/MqttClientStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttClientStatus.cs @@ -1,7 +1,7 @@ using MQTTnet.Formatter; using System; using System.Threading.Tasks; -using MQTTnet.Protocol; +using MQTTnet.Client.Disconnecting; namespace MQTTnet.Server.Status { @@ -46,7 +46,7 @@ namespace MQTTnet.Server.Status public Task DisconnectAsync() { - return _connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection); + return _connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection); } public void ResetStatistics()