diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs index 53603fa5..58ab4521 100644 --- a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs @@ -60,8 +60,11 @@ namespace MQTTnet.Adapter _logger.Trace("TX >>> {0} [Timeout={1}]", packet, timeout); - var writeBuffer = PacketSerializer.Serialize(packet); - await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false); + var chunks = PacketSerializer.Serialize(packet); + foreach (var chunk in chunks) + { + await _channel.SendStream.WriteAsync(chunk.Array, chunk.Offset, chunk.Count, cancellationToken).ConfigureAwait(false); + } } if (timeout > TimeSpan.Zero) @@ -114,7 +117,7 @@ namespace MQTTnet.Adapter { receivedMqttPacket?.Dispose(); } - }); + }).ConfigureAwait(false); return packet; } @@ -140,8 +143,8 @@ namespace MQTTnet.Adapter var readBytesCount = await stream.ReadAsync(body, offset, body.Length - offset, cancellationToken).ConfigureAwait(false); offset += readBytesCount; } while (offset < header.BodyLength); - - return new ReceivedMqttPacket(header, new MemoryStream(body, 0, body.Length)); + + return new ReceivedMqttPacket(header, new MemoryStream(body, 0, body.Length, false, true)); } private static async Task ExecuteAndWrapExceptionAsync(Func action) diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs b/Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs index 6b46b528..439460d1 100644 --- a/Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs +++ b/Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs @@ -1,4 +1,6 @@ -using MQTTnet.Adapter; +using System; +using System.Collections.Generic; +using MQTTnet.Adapter; using MQTTnet.Packets; namespace MQTTnet.Serializer @@ -7,7 +9,7 @@ namespace MQTTnet.Serializer { MqttProtocolVersion ProtocolVersion { get; set; } - byte[] Serialize(MqttBasePacket mqttPacket); + IEnumerable> Serialize(MqttBasePacket mqttPacket); MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket); } diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs index caf28813..04ee01f2 100644 --- a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs +++ b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Generic; using System.IO; using System.Text; using System.Threading; @@ -30,7 +29,7 @@ namespace MQTTnet.Serializer { return null; } - + var fixedHeader = (byte)buffer; var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4); var bodyLength = ReadBodyLengthFromSource(stream, cancellationToken); @@ -57,12 +56,22 @@ namespace MQTTnet.Serializer public string ReadStringWithLengthPrefix() { var buffer = ReadWithLengthPrefix(); + if (buffer.Length == 0) + { + return string.Empty; + } + return Encoding.UTF8.GetString(buffer, 0, buffer.Length); } public byte[] ReadWithLengthPrefix() { var length = ReadUInt16(); + if (length == 0) + { + return new byte[0]; + } + return ReadBytes(length); } @@ -78,7 +87,7 @@ namespace MQTTnet.Serializer var value = 0; byte encodedByte; - var readBytes = new List(); + ////var readBytes = new List(); do { if (cancellationToken.IsCancellationRequested) @@ -87,19 +96,20 @@ namespace MQTTnet.Serializer } var buffer = stream.ReadByte(); - readBytes.Add(buffer); - if (buffer == -1) { throw new MqttCommunicationException("Connection closed while reading remaining length data."); } + ////readBytes.Add(buffer); + encodedByte = (byte)buffer; value += (byte)(encodedByte & 127) * multiplier; multiplier *= 128; if (multiplier > 128 * 128 * 128) { - throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)})."); + //throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)})."); + throw new MqttProtocolViolationException("Remaining length is invalid."); } } while ((encodedByte & 128) != 0); diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs index ce45b3e2..0b041146 100644 --- a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs +++ b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs @@ -1,12 +1,12 @@ -using System; +using MQTTnet.Adapter; +using MQTTnet.Exceptions; +using MQTTnet.Packets; +using MQTTnet.Protocol; +using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Text; -using MQTTnet.Adapter; -using MQTTnet.Exceptions; -using MQTTnet.Packets; -using MQTTnet.Protocol; namespace MQTTnet.Serializer { @@ -17,25 +17,29 @@ namespace MQTTnet.Serializer public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; - public byte[] Serialize(MqttBasePacket packet) + public IEnumerable> Serialize(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); - using (var stream = new MemoryStream()) + using (var stream = new MemoryStream(128)) using (var writer = new MqttPacketWriter(stream)) { var fixedHeader = SerializePacket(packet, writer); - var headerBuffer = new List { fixedHeader }; - MqttPacketWriter.WriteRemainingLength((int)stream.Length, headerBuffer); + var remainingLength = (int)stream.Length; + writer.Write(fixedHeader); + MqttPacketWriter.WriteRemainingLength(remainingLength, writer); + var headerLength = (int)stream.Length - remainingLength; - var header = headerBuffer.ToArray(); - var body = stream.ToArray(); - - var buffer = new byte[header.Length + body.Length]; - Buffer.BlockCopy(header, 0, buffer, 0, header.Length); - Buffer.BlockCopy(body, 0, buffer, header.Length, body.Length); - - return buffer; +#if NET461 || NET452 || NETSTANDARD2_0 + var buffer = stream.GetBuffer(); +#else + var buffer = stream.ToArray(); +#endif + return new List> + { + new ArraySegment(buffer, remainingLength, headerLength), + new ArraySegment(buffer, 0, remainingLength) + }; } } diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs index cdf7915e..cb3d4589 100644 --- a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs +++ b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Generic; using System.IO; using System.Text; using MQTTnet.Protocol; @@ -56,11 +55,11 @@ namespace MQTTnet.Serializer Write(value); } - public static void WriteRemainingLength(int length, List target) + public static void WriteRemainingLength(int length, BinaryWriter target) { if (length == 0) { - target.Add(0); + target.Write((byte)0); return; } @@ -75,7 +74,7 @@ namespace MQTTnet.Serializer encodedByte = encodedByte | 128; } - target.Add((byte)encodedByte); + target.Write((byte)encodedByte); } while (x > 0); } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs index f871ad38..275fcb1b 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; @@ -14,8 +13,6 @@ namespace MQTTnet.Server { public sealed class MqttClientSession { - private readonly HashSet _unacknowledgedPublishPackets = new HashSet(); - private readonly MqttClientSubscriptionsManager _subscriptionsManager; private readonly MqttClientSessionsManager _sessionsManager; private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue; @@ -99,21 +96,22 @@ namespace MQTTnet.Server if (willMessage != null) { _willMessage = null; //clear willmessage so it is send just once - await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage); + await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage).ConfigureAwait(false); } } } - public void EnqueuePublishPacket(MqttPublishPacket publishPacket) + public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage) { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - var result = _subscriptionsManager.CheckSubscriptions(publishPacket); + var result = _subscriptionsManager.CheckSubscriptions(applicationMessage); if (!result.IsSubscribed) { return; } + var publishPacket = applicationMessage.ToPublishPacket(); publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel; _pendingMessagesQueue.Enqueue(publishPacket); } @@ -134,12 +132,12 @@ namespace MQTTnet.Server catch (MqttCommunicationException exception) { _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); - await StopAsync(); + await StopAsync().ConfigureAwait(false); } catch (Exception exception) { _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); - await StopAsync(); + await StopAsync().ConfigureAwait(false); } } @@ -200,7 +198,7 @@ namespace MQTTnet.Server if (subscribeResult.CloseConnection) { await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false); - await StopAsync(); + await StopAsync().ConfigureAwait(false); } } @@ -209,7 +207,7 @@ namespace MQTTnet.Server var retainedMessages = await _sessionsManager.GetRetainedMessagesAsync(subscribePacket).ConfigureAwait(false); foreach (var publishPacket in retainedMessages) { - EnqueuePublishPacket(publishPacket.ToPublishPacket()); + EnqueueApplicationMessage(publishPacket); } } @@ -221,27 +219,22 @@ namespace MQTTnet.Server { case MqttQualityOfServiceLevel.AtMostOnce: { - await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage); + await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); return; } case MqttQualityOfServiceLevel.AtLeastOnce: { - await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage); + await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, - new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); + new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false); return; } case MqttQualityOfServiceLevel.ExactlyOnce: { // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] - lock (_unacknowledgedPublishPackets) - { - _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier); - } - - await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage); + await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false); @@ -255,11 +248,6 @@ namespace MQTTnet.Server private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { - lock (_unacknowledgedPublishPackets) - { - _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier); - } - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }); } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index 7d647d92..defe5ebc 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -6,7 +6,6 @@ using System.Threading.Tasks; using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; -using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Serializer; @@ -59,7 +58,7 @@ namespace MQTTnet.Server return; } - var clientSession = await GetOrCreateClientSessionAsync(connectPacket); + var clientSession = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false); await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket { @@ -105,7 +104,7 @@ namespace MQTTnet.Server { foreach (var session in _sessions) { - await session.Value.StopAsync(); + await session.Value.StopAsync().ConfigureAwait(false); } _sessions.Clear(); @@ -162,7 +161,7 @@ namespace MQTTnet.Server { foreach (var clientSession in _sessions.Values.ToList()) { - clientSession.EnqueuePublishPacket(applicationMessage.ToPublishPacket()); + clientSession.EnqueueApplicationMessage(applicationMessage); } } } @@ -200,7 +199,7 @@ namespace MQTTnet.Server if (connectPacket.CleanSession) { _sessions.Remove(connectPacket.ClientId); - await clientSession.StopAsync(); + await clientSession.StopAsync().ConfigureAwait(false); clientSession = null; _logger.Trace("Stopped existing session of client '{0}'.", connectPacket.ClientId); diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs index 1276b066..73548ce3 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs @@ -64,23 +64,23 @@ namespace MQTTnet.Server return unsubscribePacket.CreateResponse(); } - public CheckSubscriptionsResult CheckSubscriptions(MqttPublishPacket publishPacket) + public CheckSubscriptionsResult CheckSubscriptions(MqttApplicationMessage applicationMessage) { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); lock (_subscriptions) { foreach (var subscription in _subscriptions) { - if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscription.Key)) + if (!MqttTopicFilterComparer.IsMatch(applicationMessage.Topic, subscription.Key)) { continue; } var effectiveQos = subscription.Value; - if (publishPacket.QualityOfServiceLevel < effectiveQos) + if (applicationMessage.QualityOfServiceLevel < effectiveQos) { - effectiveQos = publishPacket.QualityOfServiceLevel; + effectiveQos = applicationMessage.QualityOfServiceLevel; } return new CheckSubscriptionsResult diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index 01d19f1f..25c93520 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.IO; using System.Text; using System.Threading; @@ -392,9 +393,9 @@ namespace MQTTnet.Core.Tests private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) { var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion }; - var buffer = serializer.Serialize(packet); - - Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer)); + var chunks = serializer.Serialize(packet); + + Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(chunks))); } private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value) @@ -403,18 +404,29 @@ namespace MQTTnet.Core.Tests var buffer1 = serializer.Serialize(packet); - using (var headerStream = new MemoryStream(buffer1)) + using (var headerStream = new MemoryStream(Join(buffer1))) { var header = MqttPacketReader.ReadHeaderFromSource(headerStream, CancellationToken.None); - using (var bodyStream = new MemoryStream(buffer1, (int)headerStream.Position, header.BodyLength)) + using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.BodyLength)) { var deserializedPacket = serializer.Deserialize(new ReceivedMqttPacket(header, bodyStream)); var buffer2 = serializer.Serialize(deserializedPacket); - Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer2)); + Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2))); } } } + + private static byte[] Join(IEnumerable> chunks) + { + var buffer = new MemoryStream(); + foreach (var chunk in chunks) + { + buffer.Write(chunk.Array, chunk.Offset, chunk.Count); + } + + return buffer.ToArray(); + } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs b/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs index 2c24c483..b6b68dce 100644 --- a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs @@ -18,7 +18,7 @@ namespace MQTTnet.Core.Tests sm.Subscribe(sp, ""); - var pp = new MqttPublishPacket + var pp = new MqttApplicationMessage { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce @@ -37,7 +37,7 @@ namespace MQTTnet.Core.Tests sm.Subscribe(sp, ""); - var pp = new MqttPublishPacket + var pp = new MqttApplicationMessage { Topic = "A/B/X", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce @@ -56,7 +56,7 @@ namespace MQTTnet.Core.Tests sm.Subscribe(sp, ""); - var pp = new MqttPublishPacket + var pp = new MqttApplicationMessage { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce diff --git a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs index 3aba3c1c..301e0fec 100644 --- a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs @@ -171,16 +171,16 @@ namespace MQTTnet.TestApp.NetCore var msgs = 0; var stopwatch = Stopwatch.StartNew(); - mqttServer.ApplicationMessageReceived += (sender, args) => - { - msgs++; - if (stopwatch.ElapsedMilliseconds > 1000) - { - Console.WriteLine($"received {msgs}"); - msgs = 0; - stopwatch.Restart(); - } - }; + ////mqttServer.ApplicationMessageReceived += (sender, args) => + ////{ + //// msgs++; + //// if (stopwatch.ElapsedMilliseconds > 1000) + //// { + //// Console.WriteLine($"received {msgs}"); + //// msgs = 0; + //// stopwatch.Restart(); + //// } + ////}; await mqttServer.StartAsync(new MqttServerOptions()); Console.WriteLine("Press any key to exit.");