YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-25 07:41:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add PayloadSegmentable feature (#1585)

Browse Source
This commit is contained in:
老九
2023-02-22 20:53:55 +01:00
committed by GitHub
parent ba57537108
commit d0584413ca
24 changed files with 260 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs
+1 -1
View File
@@ -19,7 +19,7 @@ namespace MQTTnet.Formatter
return new MqttApplicationMessage
{
Topic = publishPacket.Topic,
Payload = publishPacket.Payload,
PayloadSegment = publishPacket.PayloadSegment,
QualityOfServiceLevel = publishPacket.QualityOfServiceLevel,
Retain = publishPacket.Retain,
Dup = publishPacket.Dup,
Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs
+9 -3
View File
@@ -16,7 +16,7 @@ namespace MQTTnet.Formatter
return new MqttPublishPacket
{
Topic = publishPacket.Topic,
Payload = publishPacket.Payload,
PayloadSegment = publishPacket.PayloadSegment,
Retain = publishPacket.Retain,
QualityOfServiceLevel = publishPacket.QualityOfServiceLevel,
Dup = publishPacket.Dup,
@@ -36,7 +36,7 @@ namespace MQTTnet.Formatter
var packet = new MqttPublishPacket
{
Topic = applicationMessage.Topic,
Payload = applicationMessage.Payload,
PayloadSegment = applicationMessage.PayloadSegment,
QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel,
Retain = applicationMessage.Retain,
Dup = applicationMessage.Dup,
@@ -65,10 +65,16 @@ namespace MQTTnet.Formatter
throw new MqttProtocolViolationException("The CONNECT packet contains no will message (WillFlag).");
}
ArraySegment<byte> willMessageBuffer = default;
if (connectPacket.WillMessage?.Length > 0)
{
willMessageBuffer = new ArraySegment<byte>(connectPacket.WillMessage);
}
var packet = new MqttPublishPacket
{
Topic = connectPacket.WillTopic,
Payload = connectPacket.WillMessage,
PayloadSegment = willMessageBuffer,
QualityOfServiceLevel = connectPacket.WillQoS,
Retain = connectPacket.WillRetain,
ContentType = connectPacket.WillContentType,
Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs
+13 -17
View File
@@ -54,12 +54,12 @@ namespace MQTTnet.Formatter.V3
return DecodePubRelPacket(receivedMqttPacket.Body);
case MqttControlPacketType.PubComp:
return DecodePubCompPacket(receivedMqttPacket.Body);
case MqttControlPacketType.PingReq:
return MqttPingReqPacket.Instance;
case MqttControlPacketType.PingResp:
return MqttPingRespPacket.Instance;
case MqttControlPacketType.Connect:
return DecodeConnectPacket(receivedMqttPacket.Body);
case MqttControlPacketType.ConnAck:
@@ -73,7 +73,7 @@ namespace MQTTnet.Formatter.V3
}
case MqttControlPacketType.Disconnect:
return DisconnectPacket;
case MqttControlPacketType.Subscribe:
return DecodeSubscribePacket(receivedMqttPacket.Body);
case MqttControlPacketType.SubAck:
@@ -103,9 +103,10 @@ namespace MQTTnet.Formatter.V3
var remainingLength = (uint)(_bufferWriter.Length - 5);
var publishPacket = packet as MqttPublishPacket;
if (publishPacket?.Payload != null)
var payloadSegment = publishPacket?.PayloadSegment;
if (payloadSegment != null)
{
remainingLength += (uint)publishPacket.Payload.Length;
remainingLength += (uint)payloadSegment.Value.Count;
}
var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength);
@@ -119,18 +120,13 @@ namespace MQTTnet.Formatter.V3
_bufferWriter.WriteVariableByteInteger(remainingLength);
var buffer = _bufferWriter.GetBuffer();
var firstSegment = new ArraySegment<byte>(buffer, headerOffset, _bufferWriter.Length - headerOffset);
if (publishPacket?.Payload != null)
{
var payloadSegment = new ArraySegment<byte>(publishPacket.Payload, 0, publishPacket.Payload.Length);
return new MqttPacketBuffer(firstSegment, payloadSegment);
}
return new MqttPacketBuffer(firstSegment);
return payloadSegment == null
? new MqttPacketBuffer(firstSegment)
: new MqttPacketBuffer(firstSegment, payloadSegment.Value);
}
MqttPacket DecodeConnAckPacket(ArraySegment<byte> body)
{
ThrowIfBodyIsEmpty(body);
@@ -177,7 +173,7 @@ namespace MQTTnet.Formatter.V3
var tryPrivate = (protocolVersion & 0x80) > 0;
protocolVersion &= 0x7F;
if (protocolVersion != 3 && protocolVersion != 4)
{
throw new MqttProtocolViolationException("MQTT protocol version do not match MQTT v3.");
@@ -282,7 +278,7 @@ namespace MQTTnet.Formatter.V3
if (!_bufferReader.EndOfStream)
{
packet.Payload = _bufferReader.ReadRemainingData();
packet.PayloadSegment = new ArraySegment<byte>(_bufferReader.ReadRemainingData());
}
return packet;
@@ -421,7 +417,7 @@ namespace MQTTnet.Formatter.V3
{
protocolVersion |= 0x80;
}
bufferWriter.WriteByte((byte)protocolVersion);
byte connectFlags = 0x0;
Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs
+1 -1
View File
@@ -528,7 +528,7 @@ namespace MQTTnet.Formatter.V5
if (!_bufferReader.EndOfStream)
{
packet.Payload = _bufferReader.ReadRemainingData();
packet.PayloadSegment = new ArraySegment<byte>(_bufferReader.ReadRemainingData());
}
return packet;
Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs
+8 -11
View File
@@ -37,9 +37,11 @@ namespace MQTTnet.Formatter.V5
var remainingLength = (uint)_bufferWriter.Length - 5;
var publishPacket = packet as MqttPublishPacket;
if (publishPacket?.Payload != null)
var payloadSegment = publishPacket?.PayloadSegment;
if (payloadSegment != null)
{
remainingLength += (uint)publishPacket.Payload.Length;
remainingLength += (uint)payloadSegment.Value.Count;
}
var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength);
@@ -53,18 +55,13 @@ namespace MQTTnet.Formatter.V5
_bufferWriter.WriteVariableByteInteger(remainingLength);
var buffer = _bufferWriter.GetBuffer();
var firstSegment = new ArraySegment<byte>(buffer, headerOffset, _bufferWriter.Length - headerOffset);
if (publishPacket?.Payload != null)
{
var payloadSegment = new ArraySegment<byte>(publishPacket.Payload, 0, publishPacket.Payload.Length);
return new MqttPacketBuffer(firstSegment, payloadSegment);
}
return new MqttPacketBuffer(firstSegment);
return payloadSegment == null
? new MqttPacketBuffer(firstSegment)
: new MqttPacketBuffer(firstSegment, payloadSegment.Value);
}
byte EncodeAuthPacket(MqttAuthPacket packet)
{
_bufferWriter.WriteByte((byte)packet.ReasonCode);
Source/MQTTnet/MqttApplicationMessage.cs
+52 -4
View File
@@ -2,14 +2,19 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
namespace MQTTnet
{
public sealed class MqttApplicationMessage
{
byte[] _payloadCache;
ArraySegment<byte> _payloadSegment = EmptyBuffer.ArraySegment;
/// <summary>
/// Gets or sets the content type.
/// The content type must be a UTF-8 encoded string. The content type value identifies the kind of UTF-8 encoded
@@ -48,10 +53,53 @@ namespace MQTTnet
public uint MessageExpiryInterval { get; set; }
/// <summary>
/// Gets or sets the payload.
/// The payload is the data bytes sent via the MQTT protocol.
/// Gets or sets the payload.
/// The payload is the data bytes sent via the MQTT protocol.
/// </summary>
public byte[] Payload { get; set; }
[Obsolete("Use PayloadSegment instead. This property will be removed in a future release.")]
public byte[] Payload
{
get
{
if (_payloadSegment.Array == null)
{
return null;
}
// just reference from _payloadSegment.Array
if (_payloadSegment.Count == _payloadSegment.Array.Length)
{
return _payloadSegment.Array;
}
// copy from _payloadSegment
if (_payloadCache == null)
{
_payloadCache = new byte[_payloadSegment.Count];
MqttMemoryHelper.Copy(_payloadSegment.Array, _payloadSegment.Offset, _payloadCache, 0, _payloadCache.Length);
}
return _payloadCache;
}
set
{
_payloadCache = null;
_payloadSegment = value == null || value.Length == 0 ? EmptyBuffer.ArraySegment : new ArraySegment<byte>(value);
}
}
/// <summary>
/// Get or set ArraySegment style of Payload.
/// </summary>
public ArraySegment<byte> PayloadSegment
{
get => _payloadSegment;
set
{
_payloadCache = null;
_payloadSegment = value;
}
}
/// <summary>
/// Gets or sets the payload format indicator.
Source/MQTTnet/MqttApplicationMessageBuilder.cs
+58 -43
View File
@@ -2,13 +2,16 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Runtime.InteropServices;
using System.Text;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet
{
@@ -17,7 +20,8 @@ namespace MQTTnet
string _contentType;
byte[] _correlationData;
uint _messageExpiryInterval;
byte[] _payload;
ArraySegment<byte> _payloadSegment;
MqttPayloadFormatIndicator _payloadFormatIndicator;
MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
string _responseTopic;
@@ -37,7 +41,7 @@ namespace MQTTnet
var applicationMessage = new MqttApplicationMessage
{
Topic = _topic,
Payload = _payload,
PayloadSegment = _payloadSegment,
QualityOfServiceLevel = _qualityOfServiceLevel,
Retain = _retain,
ContentType = _contentType,
@@ -85,7 +89,15 @@ namespace MQTTnet
public MqttApplicationMessageBuilder WithPayload(byte[] payload)
{
_payload = payload;
_payloadSegment = payload == null || payload.Length == 0
? EmptyBuffer.ArraySegment
: new ArraySegment<byte>(payload);
return this;
}
public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment<byte> payloadSegment)
{
_payloadSegment = payloadSegment;
return this;
}
@@ -93,70 +105,73 @@ namespace MQTTnet
{
if (payload == null)
{
_payload = null;
return this;
return WithPayload(default(byte[]));
}
_payload = payload as byte[] ?? payload.ToArray();
if (payload is byte[] byteArray)
{
return WithPayload(byteArray);
}
return this;
if (payload is ArraySegment<byte> arraySegment)
{
return WithPayloadSegment(arraySegment);
}
return WithPayload(payload.ToArray());
}
public MqttApplicationMessageBuilder WithPayload(Stream payload)
{
if (payload == null)
{
_payload = null;
return this;
}
return WithPayload(payload, payload.Length - payload.Position);
return payload == null
? WithPayload(default(byte[]))
: WithPayload(payload, payload.Length - payload.Position);
}
public MqttApplicationMessageBuilder WithPayload(Stream payload, long length)
{
if (payload == null)
if (payload == null || length == 0)
{
_payload = null;
return this;
return WithPayload(default(byte[]));
}
if (payload.Length == 0)
var payloadBuffer = new byte[length];
var totalRead = 0;
do
{
_payload = null;
}
else
{
_payload = new byte[length];
var totalRead = 0;
do
var bytesRead = payload.Read(payloadBuffer, totalRead, payloadBuffer.Length - totalRead);
if (bytesRead == 0)
{
var bytesRead = payload.Read(_payload, totalRead, _payload.Length - totalRead);
if (bytesRead == 0)
{
break;
}
break;
}
totalRead += bytesRead;
} while (totalRead < length);
}
totalRead += bytesRead;
} while (totalRead < length);
return this;
return WithPayload(payloadBuffer);
}
public MqttApplicationMessageBuilder WithPayload(string payload)
{
if (payload == null)
if (string.IsNullOrEmpty(payload))
{
_payload = null;
return this;
return WithPayload(default(byte[]));
}
_payload = string.IsNullOrEmpty(payload) ? null : Encoding.UTF8.GetBytes(payload);
return this;
var payloadBuffer = Encoding.UTF8.GetBytes(payload);
return WithPayload(payloadBuffer);
}
#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1
public MqttApplicationMessageBuilder WithPayloadSegment(ReadOnlyMemory<byte> payloadSegment)
{
return MemoryMarshal.TryGetArray(payloadSegment, out var segment)
? WithPayloadSegment(segment)
: WithPayload(payloadSegment.ToArray());
}
#endif
/// <summary>
/// Adds the payload format indicator to the message.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
Source/MQTTnet/MqttApplicationMessageExtensions.cs
+4 -10
View File
@@ -11,19 +11,13 @@ namespace MQTTnet
{
public static string ConvertPayloadToString(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
if (applicationMessage.Payload == null)
if (applicationMessage == null)
{
return null;
throw new ArgumentNullException(nameof(applicationMessage));
}
if (applicationMessage.Payload.Length == 0)
{
return string.Empty;
}
return Encoding.UTF8.GetString(applicationMessage.Payload, 0, applicationMessage.Payload.Length);
var payloadSegment = applicationMessage.PayloadSegment;
return Encoding.UTF8.GetString(payloadSegment.Array, payloadSegment.Offset, payloadSegment.Count);
}
}
}
Source/MQTTnet/Packets/MqttPublishPacket.cs
+4 -3
View File
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic;
using MQTTnet.Protocol;
@@ -17,10 +18,10 @@ namespace MQTTnet.Packets
public uint MessageExpiryInterval { get; set; }
public byte[] Payload { get; set; }
public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified;
public ArraySegment<byte> PayloadSegment { get; set; }
public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; } = MqttQualityOfServiceLevel.AtMostOnce;
public string ResponseTopic { get; set; }
@@ -38,7 +39,7 @@ namespace MQTTnet.Packets
public override string ToString()
{
return
$"Publish: [Topic={Topic}] [Payload.Length={Payload?.Length}] [QoSLevel={QualityOfServiceLevel}] [Dup={Dup}] [Retain={Retain}] [PacketIdentifier={PacketIdentifier}]";
$"Publish: [Topic={Topic}] [PayloadLength={PayloadSegment.Count}] [QoSLevel={QualityOfServiceLevel}] [Dup={Dup}] [Retain={Retain}] [PacketIdentifier={PacketIdentifier}]";
}
}
}
Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs
+13 -3
View File
@@ -25,6 +25,7 @@ namespace MQTTnet.Server
if (logger == null)
throw new ArgumentNullException(nameof(logger));
_logger = logger.WithSource(nameof(MqttRetainedMessagesManager));
}
@@ -68,7 +69,8 @@ namespace MQTTnet.Server
lock (_messages)
{
var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
var payloadSegment = applicationMessage.PayloadSegment;
var hasPayload = payloadSegment.Count > 0;
if (!hasPayload)
{
@@ -84,8 +86,7 @@ namespace MQTTnet.Server
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel ||
!existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? EmptyBuffer.Array))
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !SequenceEqual(existingMessage.PayloadSegment, payloadSegment))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
@@ -150,5 +151,14 @@ namespace MQTTnet.Server
await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
}
}
private static bool SequenceEqual(ArraySegment<byte> source, ArraySegment<byte> target)
{
#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1
return source.AsSpan().SequenceEqual(target);
#else
return source.Count == target.Count && Enumerable.SequenceEqual(source, target);
#endif
}
}
}
Source/MQTTnet/Server/MqttServerExtensions.cs
+1 -1
View File
@@ -41,7 +41,7 @@ namespace MQTTnet.Server
new MqttApplicationMessage
{
Topic = topic,
Payload = payloadBuffer,
PayloadSegment = new ArraySegment<byte>(payloadBuffer),
QualityOfServiceLevel = qualityOfServiceLevel,
Retain = retain
}));