Use the data converter only when creating packets.

This commit is contained in:
Christian
2021-02-19 21:45:23 +01:00
parent cfc4ebe6a3
commit ed520455d0
12 changed files with 153 additions and 96 deletions
@@ -12,7 +12,7 @@
ServerBusy = 137, ServerBusy = 137,
ServerShuttingDown = 139, ServerShuttingDown = 139,
BadAuthenticationMethod = 140, BadAuthenticationMethod = 140,
KeepaliveTimeout = 141, KeepAliveTimeout = 141,
SessionTakenOver = 142, SessionTakenOver = 142,
TopicFilterInvalid = 143, TopicFilterInvalid = 143,
TopicNameInvalid = 144, TopicNameInvalid = 144,
+13 -29
View File
@@ -167,7 +167,7 @@ namespace MQTTnet.Client
public Task PingAsync(CancellationToken cancellationToken) public Task PingAsync(CancellationToken cancellationToken)
{ {
return SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken); return SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken);
} }
public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken) public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken)
@@ -650,11 +650,8 @@ namespace MQTTnet.Client
if (!eventArgs.ProcessingFailed) if (!eventArgs.ProcessingFailed)
{ {
await SendAsync(new MqttPubAckPacket var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, eventArgs.ReasonCode);
{ await SendAsync(pubAckPacket, cancellationToken).ConfigureAwait(false);
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = (MqttPubAckReasonCode)eventArgs.ReasonCode
}, cancellationToken).ConfigureAwait(false);
} }
} }
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -663,11 +660,8 @@ namespace MQTTnet.Client
if (!eventArgs.ProcessingFailed) if (!eventArgs.ProcessingFailed)
{ {
await SendAsync(new MqttPubRecPacket var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, eventArgs.ReasonCode);
{ await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = (MqttPubRecReasonCode)eventArgs.ReasonCode
}, cancellationToken).ConfigureAwait(false);
} }
} }
else else
@@ -691,11 +685,8 @@ namespace MQTTnet.Client
{ {
// The packet is unknown. Probably due to a restart of the 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. // So wen send this to the server to trigger a full resend of the message.
return SendAsync(new MqttPubRelPacket var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
{ return SendAsync(pubRelPacket, cancellationToken);
PacketIdentifier = pubRecPacket.PacketIdentifier,
ReasonCode = MqttPubRelReasonCode.PacketIdentifierNotFound
}, cancellationToken);
} }
return PlatformAbstractionLayer.CompletedTask; return PlatformAbstractionLayer.CompletedTask;
@@ -703,11 +694,8 @@ namespace MQTTnet.Client
Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{ {
return SendAsync(new MqttPubCompPacket var pubCompPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
{ return SendAsync(pubCompPacket, cancellationToken);
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
}, cancellationToken);
} }
Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) 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] // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false); await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false);
return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(null); return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(null);
} }
async Task<MqttClientPublishResult> PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) async Task<MqttClientPublishResult> PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{ {
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var response = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false); var response = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(response); return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(response);
} }
async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -756,15 +744,11 @@ namespace MQTTnet.Client
var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false); var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
var pubRelPacket = new MqttPubRelPacket var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRelReasonCode.Success
};
var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false); var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket); return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubRecPacket, pubCompPacket);
} }
async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
+16 -12
View File
@@ -14,23 +14,31 @@ namespace MQTTnet.Formatter
{ {
public interface IMqttDataConverter public interface IMqttDataConverter
{ {
MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage);
MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket);
MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket);
MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket); MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket);
MqttClientAuthenticateResult CreateClientConnectResult(MqttConnAckPacket connAckPacket); 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); MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, IMqttClientOptions options);
MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext); MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext);
MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket); MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage);
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); MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options);
@@ -41,9 +49,5 @@ namespace MQTTnet.Formatter
MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes); MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes);
MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options); MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options);
MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket);
MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket);
} }
} }
@@ -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 return new MqttPubAckPacket
{ {
PacketIdentifier = publishPacket.PacketIdentifier, PacketIdentifier = publishPacket.PacketIdentifier
ReasonCode = MqttPubAckReasonCode.Success
}; };
} }
public MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket) public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
{ {
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
return new MqttPubRecPacket return new MqttPubRecPacket
{ {
PacketIdentifier = publishPacket.PacketIdentifier, PacketIdentifier = publishPacket.PacketIdentifier
ReasonCode = MqttPubRecReasonCode.Success };
}
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, Topic = publishPacket.Topic,
Payload = publishPacket.Payload, Payload = publishPacket.Payload,
QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, 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) 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 var subackPacket = new MqttSubAckPacket
{ {
PacketIdentifier = subscribePacket.PacketIdentifier PacketIdentifier = subscribePacket.PacketIdentifier
@@ -206,6 +232,9 @@ namespace MQTTnet.Formatter.V3
public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes) public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes)
{ {
if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes));
return new MqttUnsubAckPacket return new MqttUnsubAckPacket
{ {
PacketIdentifier = unsubscribePacket.PacketIdentifier, PacketIdentifier = unsubscribePacket.PacketIdentifier,
@@ -223,7 +252,7 @@ namespace MQTTnet.Formatter.V3
return new MqttDisconnectPacket(); return new MqttDisconnectPacket();
} }
public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket) public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket)
{ {
return new MqttClientPublishResult 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) if (pubRecPacket == null || pubCompPacket == null)
{ {
@@ -15,7 +15,7 @@ using MqttClientSubscribeResult = MQTTnet.Client.Subscribing.MqttClientSubscribe
namespace MQTTnet.Formatter.V5 namespace MQTTnet.Formatter.V5
{ {
public class MqttV500DataConverter : IMqttDataConverter public sealed class MqttV500DataConverter : IMqttDataConverter
{ {
public MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage) public MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage)
{ {
@@ -27,7 +27,7 @@ namespace MQTTnet.Formatter.V5
Payload = applicationMessage.Payload, Payload = applicationMessage.Payload,
QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel,
Retain = applicationMessage.Retain, Retain = applicationMessage.Retain,
Dup = false, Dup = applicationMessage.Dup,
Properties = new MqttPublishPacketProperties Properties = new MqttPublishPacketProperties
{ {
ContentType = applicationMessage.ContentType, ContentType = applicationMessage.ContentType,
@@ -49,32 +49,61 @@ namespace MQTTnet.Formatter.V5
return packet; 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 return new MqttPubAckPacket
{ {
PacketIdentifier = publishPacket.PacketIdentifier, 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 return new MqttPubRecPacket
{ {
PacketIdentifier = publishPacket.PacketIdentifier, 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) public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket)
{ {
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
return new MqttApplicationMessage return new MqttApplicationMessage
{ {
Topic = publishPacket.Topic, Topic = publishPacket.Topic,
Payload = publishPacket.Payload, Payload = publishPacket.Payload,
QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, QualityOfServiceLevel = publishPacket.QualityOfServiceLevel,
Retain = publishPacket.Retain, Retain = publishPacket.Retain,
Dup = publishPacket.Dup,
ResponseTopic = publishPacket.Properties?.ResponseTopic, ResponseTopic = publishPacket.Properties?.ResponseTopic,
ContentType = publishPacket.Properties?.ContentType, ContentType = publishPacket.Properties?.ContentType,
CorrelationData = publishPacket.Properties?.CorrelationData, CorrelationData = publishPacket.Properties?.CorrelationData,
@@ -93,7 +122,7 @@ namespace MQTTnet.Formatter.V5
return new MqttClientAuthenticateResult return new MqttClientAuthenticateResult
{ {
IsSessionPresent = connAckPacket.IsSessionPresent, IsSessionPresent = connAckPacket.IsSessionPresent,
ResultCode = (MqttClientConnectResultCode)connAckPacket.ReasonCode.Value, ResultCode = (MqttClientConnectResultCode)(int)(connAckPacket.ReasonCode ?? 0),
WildcardSubscriptionAvailable = connAckPacket.Properties?.WildcardSubscriptionAvailable, WildcardSubscriptionAvailable = connAckPacket.Properties?.WildcardSubscriptionAvailable,
RetainAvailable = connAckPacket.Properties?.RetainAvailable, RetainAvailable = connAckPacket.Properties?.RetainAvailable,
AssignedClientIdentifier = connAckPacket.Properties?.AssignedClientIdentifier, AssignedClientIdentifier = connAckPacket.Properties?.AssignedClientIdentifier,
@@ -144,6 +173,8 @@ namespace MQTTnet.Formatter.V5
public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext) public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext)
{ {
if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext));
return new MqttConnAckPacket return new MqttConnAckPacket
{ {
ReasonCode = connectionValidatorContext.ReasonCode, ReasonCode = connectionValidatorContext.ReasonCode,
@@ -213,6 +244,9 @@ namespace MQTTnet.Formatter.V5
public MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, Server.MqttClientSubscribeResult subscribeResult) 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 var subackPacket = new MqttSubAckPacket
{ {
PacketIdentifier = subscribePacket.PacketIdentifier PacketIdentifier = subscribePacket.PacketIdentifier
@@ -240,6 +274,9 @@ namespace MQTTnet.Formatter.V5
public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes) public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes)
{ {
if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes));
return new MqttUnsubAckPacket return new MqttUnsubAckPacket
{ {
PacketIdentifier = unsubscribePacket.PacketIdentifier, PacketIdentifier = unsubscribePacket.PacketIdentifier,
@@ -263,7 +300,7 @@ namespace MQTTnet.Formatter.V5
return packet; return packet;
} }
public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket) public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket)
{ {
var result = new MqttClientPublishResult 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. // 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. // 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; result.PacketIdentifier = pubAckPacket.PacketIdentifier;
} }
@@ -284,7 +321,7 @@ namespace MQTTnet.Formatter.V5
return result; return result;
} }
public MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
{ {
if (pubRecPacket == null || pubCompPacket == null) if (pubRecPacket == null || pubCompPacket == null)
{ {
@@ -317,7 +354,7 @@ namespace MQTTnet.Formatter.V5
if (pubRecPacket.ReasonCode.HasValue) if (pubRecPacket.ReasonCode.HasValue)
{ {
// Both enums share the same values. // Both enums share the same values.
result.ReasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode.Value; result.ReasonCode = (MqttClientPublishReasonCode)(pubRecPacket.ReasonCode ?? 0);
} }
return result; return result;
@@ -9,6 +9,7 @@
NotAuthorized = 135, NotAuthorized = 135,
TopicNameInvalid = 144, TopicNameInvalid = 144,
PacketIdentifierInUse = 145, PacketIdentifierInUse = 145,
PacketIdentifierNotFound = 146,
QuotaExceeded = 151, QuotaExceeded = 151,
PayloadFormatInvalid = 153 PayloadFormatInvalid = 153
} }
+1 -1
View File
@@ -3,7 +3,7 @@
public sealed class MqttPingReqPacket : MqttBasePacket public sealed class MqttPingReqPacket : MqttBasePacket
{ {
// This is a minor performance improvement. // This is a minor performance improvement.
public static MqttPingReqPacket Instance = new MqttPingReqPacket(); public static readonly MqttPingReqPacket Instance = new MqttPingReqPacket();
public override string ToString() public override string ToString()
{ {
+1 -1
View File
@@ -3,7 +3,7 @@
public sealed class MqttPingRespPacket : MqttBasePacket public sealed class MqttPingRespPacket : MqttBasePacket
{ {
// This is a minor performance improvement. // This is a minor performance improvement.
public static MqttPingRespPacket Instance = new MqttPingRespPacket(); public static readonly MqttPingRespPacket Instance = new MqttPingRespPacket();
public override string ToString() public override string ToString()
{ {
+22 -21
View File
@@ -13,6 +13,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Client.Disconnecting;
namespace MQTTnet.Server 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 _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere.
long _receivedApplicationMessagesCount; long _receivedApplicationMessagesCount;
long _sentApplicationMessagesCount; long _sentApplicationMessagesCount;
MqttDisconnectReasonCode _disconnectReason; MqttClientDisconnectReason _disconnectReason;
public MqttClientConnection(MqttConnectPacket connectPacket, public MqttClientConnection(MqttConnectPacket connectPacket,
IMqttChannelAdapter channelAdapter, IMqttChannelAdapter channelAdapter,
@@ -84,12 +85,12 @@ namespace MQTTnet.Server
public MqttClientSession Session { get; } public MqttClientSession Session { get; }
public async Task StopAsync(MqttDisconnectReasonCode reason) public async Task StopAsync(MqttClientDisconnectReason reason)
{ {
Status = MqttClientConnectionStatus.Finalizing; Status = MqttClientConnectionStatus.Finalizing;
_disconnectReason = reason; _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 // 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 // token because the entire connection is closed (disposed) as soon as the cancellation
@@ -97,10 +98,18 @@ namespace MQTTnet.Server
// at the client! // at the client!
try try
{ {
await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket var disconnectOptions = new MqttClientDisconnectOptions
{ {
ReasonCode = reason ReasonCode = reason,
}, CancellationToken.None).ConfigureAwait(false); 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) catch (Exception exception)
{ {
@@ -235,7 +244,7 @@ namespace MQTTnet.Server
} }
finally finally
{ {
if (_disconnectReason == MqttDisconnectReasonCode.SessionTakenOver) if (_disconnectReason == MqttClientDisconnectReason.SessionTakenOver)
{ {
disconnectType = MqttClientDisconnectType.Takeover; disconnectType = MqttClientDisconnectType.Takeover;
} }
@@ -277,12 +286,7 @@ namespace MQTTnet.Server
Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{ {
var pubCompPacket = new MqttPubCompPacket var pubCompPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
{
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
};
return SendAsync(pubCompPacket, cancellationToken); return SendAsync(pubCompPacket, cancellationToken);
} }
@@ -327,12 +331,12 @@ namespace MQTTnet.Server
} }
case MqttQualityOfServiceLevel.AtLeastOnce: case MqttQualityOfServiceLevel.AtLeastOnce:
{ {
var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket); var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
return SendAsync(pubAckPacket, cancellationToken); return SendAsync(pubAckPacket, cancellationToken);
} }
case MqttQualityOfServiceLevel.ExactlyOnce: case MqttQualityOfServiceLevel.ExactlyOnce:
{ {
var pubRecPacket = _dataConverter.CreatePubRecPacket(publishPacket); var pubRecPacket = _dataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
return SendAsync(pubRecPacket, cancellationToken); return SendAsync(pubRecPacket, cancellationToken);
} }
default: default:
@@ -441,13 +445,10 @@ namespace MQTTnet.Server
using (var awaiter2 = _packetDispatcher.AddAwaiter<MqttPubCompPacket>(publishPacket.PacketIdentifier)) using (var awaiter2 = _packetDispatcher.AddAwaiter<MqttPubCompPacket>(publishPacket.PacketIdentifier))
{ {
await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false); 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 var pubRelPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
{ await SendAsync(pubRelPacket, cancellationToken).ConfigureAwait(false);
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRelReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
} }
@@ -12,6 +12,7 @@ using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Client.Disconnecting;
using OperationCanceledException = System.OperationCanceledException; using OperationCanceledException = System.OperationCanceledException;
namespace MQTTnet.Server namespace MQTTnet.Server
@@ -121,7 +122,7 @@ namespace MQTTnet.Server
foreach (var connection in connections) 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) 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); _logger.Verbose("Session for client '{0}' deleted.", clientId);
@@ -301,7 +302,7 @@ namespace MQTTnet.Server
{ {
if (clientConnection != null) 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; _connections[connectPacket.ClientId] = connection;
} }
existingConnection?.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).GetAwaiter().GetResult(); existingConnection?.StopAsync(MqttClientDisconnectReason.SessionTakenOver).GetAwaiter().GetResult();
return connection; return connection;
} }
@@ -3,8 +3,8 @@ using MQTTnet.Internal;
using System; using System;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Implementations; using MQTTnet.Implementations;
using MQTTnet.Protocol;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -108,7 +108,7 @@ namespace MQTTnet.Server
// We do not need to wait for the task so no await is needed. // 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 // Also the internal state of the connection must be swapped to "Finalizing" because the
// next iteration of the keep alive timer happens. // next iteration of the keep alive timer happens.
var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout); var _ = connection.StopAsync(MqttClientDisconnectReason.KeepAliveTimeout);
} }
catch (Exception exception) catch (Exception exception)
{ {
@@ -1,7 +1,7 @@
using MQTTnet.Formatter; using MQTTnet.Formatter;
using System; using System;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Protocol; using MQTTnet.Client.Disconnecting;
namespace MQTTnet.Server.Status namespace MQTTnet.Server.Status
{ {
@@ -46,7 +46,7 @@ namespace MQTTnet.Server.Status
public Task DisconnectAsync() public Task DisconnectAsync()
{ {
return _connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection); return _connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection);
} }
public void ResetStatistics() public void ResetStatistics()