YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-17 09:43:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

Start migration of server to new packet bus.

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