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

Create more factories for packets.

Browse Source
This commit is contained in:
Christian
2021-10-23 10:58:21 +02:00
parent 47bc8b4d31
commit 5974e2644e
12 changed files with 134 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+4 -2
View File
@@ -680,7 +680,8 @@ namespace MQTTnet.Client
{
if (!eventArgs.ProcessingFailed)
{
var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
var pubAckPacketFactory = new MqttPubAckPacketFactory();
var pubAckPacket = pubAckPacketFactory.Create(eventArgs);
return SendAsync(pubAckPacket, cancellationToken);
}
}
@@ -688,7 +689,8 @@ namespace MQTTnet.Client
{
if (!eventArgs.ProcessingFailed)
{
var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
var pubRecPacketFactory = new MqttPubRecPacketFactory();
var pubRecPacket = pubRecPacketFactory.Create(eventArgs);
return SendAsync(pubRecPacket, cancellationToken);
}
}
Source/MQTTnet/Formatter/IMqttDataConverter.cs
-7
View File
@@ -3,7 +3,6 @@ using MQTTnet.Client.Publishing;
using MQTTnet.Client.Subscribing;
using MQTTnet.Client.Unsubscribing;
using MQTTnet.Packets;
using MQTTnet.Server;
using MQTTnet.Server.Internal;
namespace MQTTnet.Formatter
@@ -20,12 +19,6 @@ namespace MQTTnet.Formatter
MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket);
MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext);
MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs
+41
View File
@@ -0,0 +1,41 @@
using System;
using MQTTnet.Packets;
using MQTTnet.Server;
namespace MQTTnet.Formatter
{
public sealed class MqttConnAckPacketFactory
{
public MqttConnAckPacket Create(MqttConnectionValidatorContext connectionValidatorContext)
{
if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext));
var connAckPacket = new MqttConnAckPacket
{
ReturnCode = MqttConnectReasonCodeConverter.ToConnectReturnCode(connectionValidatorContext.ReasonCode),
ReasonCode = connectionValidatorContext.ReasonCode,
Properties =
{
RetainAvailable = true,
SubscriptionIdentifiersAvailable = true,
SharedSubscriptionAvailable = false,
TopicAliasMaximum = ushort.MaxValue,
WildcardSubscriptionAvailable = true,
AuthenticationMethod = connectionValidatorContext.AuthenticationMethod,
AuthenticationData = connectionValidatorContext.ResponseAuthenticationData,
AssignedClientIdentifier = connectionValidatorContext.AssignedClientIdentifier,
ReasonString = connectionValidatorContext.ReasonString,
ServerReference = connectionValidatorContext.ServerReference
}
};
if (connectionValidatorContext.ResponseUserProperties != null)
{
connAckPacket.Properties.UserProperties.AddRange(connectionValidatorContext.ResponseUserProperties);
}
return connAckPacket;
}
}
}
Source/MQTTnet/Protocol/MqttConnectReasonCodeConverter.cs → Source/MQTTnet/Formatter/MqttConnectReasonCodeConverter.cs
+5 -4
View File
@@ -1,10 +1,11 @@
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
namespace MQTTnet.Protocol
namespace MQTTnet.Formatter
{
public class MqttConnectReasonCodeConverter
public static class MqttConnectReasonCodeConverter
{
public MqttConnectReturnCode ToConnectReturnCode(MqttConnectReasonCode reasonCode)
public static MqttConnectReturnCode ToConnectReturnCode(MqttConnectReasonCode reasonCode)
{
switch (reasonCode)
{
@@ -47,7 +48,7 @@ namespace MQTTnet.Protocol
}
}
public MqttConnectReasonCode ToConnectReasonCode(MqttConnectReturnCode returnCode)
public static MqttConnectReasonCode ToConnectReasonCode(MqttConnectReturnCode returnCode)
{
switch (returnCode)
{
Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs
+32
View File
@@ -0,0 +1,32 @@
using System;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Formatter
{
public sealed class MqttPubAckPacketFactory
{
public MqttPubAckPacket Create(MqttApplicationMessageReceivedEventArgs applicationMessageReceivedEventArgs)
{
if (applicationMessageReceivedEventArgs == null) throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs));
var pubAckPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode);
pubAckPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties);
return pubAckPacket;
}
public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
var pubAckPacket = new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = (MqttPubAckReasonCode) (int) applicationMessageReceivedReasonCode
};
return pubAckPacket;
}
}
}
Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs
+35
View File
@@ -0,0 +1,35 @@
using System;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Formatter
{
public sealed class MqttPubRecPacketFactory
{
public MqttPubRecPacket Create(MqttApplicationMessageReceivedEventArgs applicationMessageReceivedEventArgs)
{
if (applicationMessageReceivedEventArgs == null) throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs));
var pubRecPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode);
pubRecPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties);
return pubRecPacket;
}
public MqttPubRecPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
var pubRecPacket = new MqttPubRecPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = (MqttPubRecReasonCode) (int) applicationMessageReceivedReasonCode
};
return pubRecPacket;
}
}
}
Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs
-30
View File
@@ -14,26 +14,6 @@ namespace MQTTnet.Formatter.V3
{
public class MqttV310DataConverter : IMqttDataConverter
{
public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
return new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier
};
}
public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
return new MqttPubRecPacket
{
PacketIdentifier = publishPacket.PacketIdentifier
};
}
public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
{
if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket));
@@ -110,16 +90,6 @@ namespace MQTTnet.Formatter.V3
};
}
public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext)
{
if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext));
return new MqttConnAckPacket
{
ReturnCode = new MqttConnectReasonCodeConverter().ToConnectReturnCode(connectionValidatorContext.ReasonCode)
};
}
public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket)
{
if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs
-56
View File
@@ -5,7 +5,6 @@ using MQTTnet.Client.Unsubscribing;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
@@ -15,30 +14,6 @@ namespace MQTTnet.Formatter.V5
{
public sealed class MqttV500DataConverter : IMqttDataConverter
{
public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket,
MqttApplicationMessageReceivedReasonCode reasonCode)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
return new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = (MqttPubAckReasonCode) (int) reasonCode
};
}
public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket,
MqttApplicationMessageReceivedReasonCode reasonCode)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
return new MqttPubRecPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = (MqttPubRecReasonCode) (int) reasonCode
};
}
public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket,
MqttApplicationMessageReceivedReasonCode reasonCode)
{
@@ -91,37 +66,6 @@ namespace MQTTnet.Formatter.V5
};
}
public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext)
{
if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext));
var connAckPacket = new MqttConnAckPacket
{
ReasonCode = connectionValidatorContext.ReasonCode,
Properties =
{
RetainAvailable = true,
SubscriptionIdentifiersAvailable = true,
SharedSubscriptionAvailable = false,
TopicAliasMaximum = ushort.MaxValue,
WildcardSubscriptionAvailable = true,
AuthenticationMethod = connectionValidatorContext.AuthenticationMethod,
AuthenticationData = connectionValidatorContext.ResponseAuthenticationData,
AssignedClientIdentifier = connectionValidatorContext.AssignedClientIdentifier,
ReasonString = connectionValidatorContext.ReasonString,
ServerReference = connectionValidatorContext.ServerReference
}
};
if (connectionValidatorContext.ResponseUserProperties != null)
{
connAckPacket.Properties.UserProperties.AddRange(connectionValidatorContext.ResponseUserProperties);
}
return connAckPacket;
}
public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket,
MqttSubAckPacket subAckPacket)
{
Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs
+7 -1
View File
@@ -1,4 +1,5 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Packets;
@@ -36,7 +37,12 @@ namespace MQTTnet
public bool ProcessingFailed { get; set; }
public MqttApplicationMessageReceivedReasonCode ReasonCode { get; set; } = MqttApplicationMessageReceivedReasonCode.Success;
/// <summary>
/// Gets or sets the user properties which being sent to the server in the ACK packet etc.
/// </summary>
public List<MqttUserProperty> ResponseUserProperties { get; } = new List<MqttUserProperty>();
/// <summary>
/// Gets or sets whether this message was handled.
/// This value can be used in user code for custom control flow.
Source/MQTTnet/Server/Internal/MqttClientConnection.cs
+4 -2
View File
@@ -449,7 +449,8 @@ namespace MQTTnet.Server.Internal
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
var pubAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
var pubAckPacketFactory = new MqttPubAckPacketFactory();
var pubAckPacket = pubAckPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket));
//return SendPacketAsync(pubAckPacket, cancellationToken);
break;
@@ -457,7 +458,8 @@ namespace MQTTnet.Server.Internal
case MqttQualityOfServiceLevel.ExactlyOnce:
{
// TODO: Migrate to packet bus.
var pubRecPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
var pubRecPacketFactory = new MqttPubRecPacketFactory();
var pubRecPacket = pubRecPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket));
//return SendPacketAsync(pubRecPacket, cancellationToken);
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+4 -4
View File
@@ -9,8 +9,6 @@ using MQTTnet.Client.Disconnecting;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Formatter.V3;
using MQTTnet.Formatter.V5;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -28,6 +26,8 @@ namespace MQTTnet.Server.Internal
readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
readonly MqttConnAckPacketFactory _connAckPacketFactory = new MqttConnAckPacketFactory();
readonly MqttServerEventDispatcher _eventDispatcher;
readonly IMqttRetainedMessagesManager _retainedMessagesManager;
@@ -102,14 +102,14 @@ namespace MQTTnet.Server.Internal
if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success)
{
// Send failure response here without preparing a session!
connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext);
connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext);
await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
return;
}
clientConnection = await CreateClientConnection(connectPacket, channelAdapter, connectionValidatorContext.SessionItems).ConfigureAwait(false);
connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext);
connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext);
await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket, channelAdapter).ConfigureAwait(false);
Source/MQTTnet/Server/MqttConnectionValidatorContext.cs
+2 -2
View File
@@ -136,8 +136,8 @@ namespace MQTTnet.Server
[Obsolete("Use ReasonCode instead. It is MQTTv5 only but will be converted to a valid ReturnCode.")]
public MqttConnectReturnCode ReturnCode
{
get => new MqttConnectReasonCodeConverter().ToConnectReturnCode(ReasonCode);
set => ReasonCode = new MqttConnectReasonCodeConverter().ToConnectReasonCode(value);
get => MqttConnectReasonCodeConverter.ToConnectReturnCode(ReasonCode);
set => ReasonCode = MqttConnectReasonCodeConverter.ToConnectReasonCode(value);
}
/// <summary>