From 49da66dcffe5e3421b0175f34491ae27163852bf Mon Sep 17 00:00:00 2001 From: Christian Date: Tue, 13 Mar 2018 23:28:50 +0100 Subject: [PATCH 1/8] Fix issue when disconnecting on UWP --- .../Implementations/MqttTcpChannel.Uwp.cs | 6 ------ 1 file changed, 6 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs index fa96a8f8..c65f789f 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs @@ -69,12 +69,6 @@ namespace MQTTnet.Implementations public void Dispose() { - SendStream?.Dispose(); - SendStream = null; - - ReceiveStream?.Dispose(); - ReceiveStream = null; - _socket?.Dispose(); _socket = null; } From 45b4c058675260e12a46f332f2f16c476187d6b5 Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 25 Mar 2018 14:38:16 +0200 Subject: [PATCH 2/8] Update version --- .../MQTTnet.TestApp.UniversalWindows.csproj | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj index 0e7656f2..b5d2dbd9 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj @@ -12,7 +12,7 @@ en-US UAP 10.0.16299.0 - 10.0.15063.0 + 10.0.16299.0 14 512 {A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC} From b3eaeea8a4dc83add1865e2b2b5831e32ce643c6 Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 25 Mar 2018 14:38:36 +0200 Subject: [PATCH 3/8] Fix log issue in UWP test app. --- Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml | 2 +- .../MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs | 10 ++++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml index dcdee000..3698613f 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml @@ -148,7 +148,7 @@ - + diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index a48b7ee5..3ed12bf4 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -42,7 +42,8 @@ namespace MQTTnet.TestApp.UniversalWindows foreach (var traceMessage in _traceMessages) { logText.AppendFormat( - "[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}", traceMessage.Timestamp, + "[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}", + traceMessage.Timestamp, traceMessage.Level, traceMessage.Source, traceMessage.ThreadId, @@ -197,8 +198,13 @@ namespace MQTTnet.TestApp.UniversalWindows } } - private void Clear(object sender, RoutedEventArgs e) + private void ClearLog(object sender, RoutedEventArgs e) { + while (_traceMessages.Count > 0) + { + _traceMessages.TryDequeue(out _); + } + Trace.Text = string.Empty; } From 447b2aa2c682dffe73ce8e7a4f790ad2d2b02996 Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 25 Mar 2018 14:39:44 +0200 Subject: [PATCH 4/8] Add new extension methods. --- .../ApplicationMessagePublisherExtensions.cs | 31 +++++++++++++++++-- .../Client/MqttClientExtensions.cs | 8 +++++ .../ManagedMqttClientOptionsBuilder.cs | 31 +++++++++++++++++-- .../MqttApplicationMessageExtensions.cs | 25 +++++++++++++++ 4 files changed, 89 insertions(+), 6 deletions(-) create mode 100644 Frameworks/MQTTnet.NetStandard/MqttApplicationMessageExtensions.cs diff --git a/Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs b/Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs index 73328de9..b1bf62f5 100644 --- a/Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs +++ b/Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs @@ -1,16 +1,41 @@ using System; using System.Threading.Tasks; +using MQTTnet.Protocol; namespace MQTTnet { public static class ApplicationMessagePublisherExtensions { - public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages) + public static Task PublishAsync(this IApplicationMessagePublisher publisher, params MqttApplicationMessage[] applicationMessages) { - if (client == null) throw new ArgumentNullException(nameof(client)); + if (publisher == null) throw new ArgumentNullException(nameof(publisher)); if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); - return client.PublishAsync(applicationMessages); + return publisher.PublishAsync(applicationMessages); + } + + public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic) + { + if (publisher == null) throw new ArgumentNullException(nameof(publisher)); + if (topic == null) throw new ArgumentNullException(nameof(topic)); + + return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).Build()); + } + + public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload) + { + if (publisher == null) throw new ArgumentNullException(nameof(publisher)); + if (topic == null) throw new ArgumentNullException(nameof(topic)); + + return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(payload).Build()); + } + + public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel) + { + if (publisher == null) throw new ArgumentNullException(nameof(publisher)); + if (topic == null) throw new ArgumentNullException(nameof(topic)); + + return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(payload).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientExtensions.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientExtensions.cs index 73f4d40c..a7548dee 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientExtensions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientExtensions.cs @@ -24,6 +24,14 @@ namespace MQTTnet.Client return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); } + public static Task> SubscribeAsync(this IMqttClient client, string topic) + { + if (client == null) throw new ArgumentNullException(nameof(client)); + if (topic == null) throw new ArgumentNullException(nameof(topic)); + + return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build()); + } + public static Task UnsubscribeAsync(this IMqttClient client, params string[] topicFilters) { if (client == null) throw new ArgumentNullException(nameof(client)); diff --git a/Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientOptionsBuilder.cs b/Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientOptionsBuilder.cs index 7c753064..38aadcfe 100644 --- a/Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientOptionsBuilder.cs +++ b/Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientOptionsBuilder.cs @@ -6,6 +6,7 @@ namespace MQTTnet.ManagedClient public class ManagedMqttClientOptionsBuilder { private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions(); + private MqttClientOptionsBuilder _clientOptionsBuilder; public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value) { @@ -21,7 +22,24 @@ namespace MQTTnet.ManagedClient public ManagedMqttClientOptionsBuilder WithClientOptions(IMqttClientOptions value) { + if (_clientOptionsBuilder != null) + { + throw new InvalidOperationException("Cannot use client options builder and client options at the same time."); + } + _options.ClientOptions = value ?? throw new ArgumentNullException(nameof(value)); + + return this; + } + + public ManagedMqttClientOptionsBuilder WithClientOptions(MqttClientOptionsBuilder builder) + { + if (_options.ClientOptions != null) + { + throw new InvalidOperationException("Cannot use client options builder and client options at the same time."); + } + + _clientOptionsBuilder = builder; return this; } @@ -29,15 +47,22 @@ namespace MQTTnet.ManagedClient { if (options == null) throw new ArgumentNullException(nameof(options)); - var builder = new MqttClientOptionsBuilder(); - options(builder); - _options.ClientOptions = builder.Build(); + if (_clientOptionsBuilder != null) + { + _clientOptionsBuilder = new MqttClientOptionsBuilder(); + } + options(_clientOptionsBuilder); return this; } public ManagedMqttClientOptions Build() { + if (_clientOptionsBuilder != null) + { + _options.ClientOptions = _clientOptionsBuilder.Build(); + } + if (_options.ClientOptions == null) { throw new InvalidOperationException("The ClientOptions cannot be null."); diff --git a/Frameworks/MQTTnet.NetStandard/MqttApplicationMessageExtensions.cs b/Frameworks/MQTTnet.NetStandard/MqttApplicationMessageExtensions.cs new file mode 100644 index 00000000..dac9998a --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/MqttApplicationMessageExtensions.cs @@ -0,0 +1,25 @@ +using System; +using System.Text; + +namespace MQTTnet +{ + public static class MqttApplicationMessageExtensions + { + public static string ConvertPayloadToString(this MqttApplicationMessage applicationMessage) + { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + + if (applicationMessage.Payload == null) + { + return null; + } + + if (applicationMessage.Payload.Length == 0) + { + return string.Empty; + } + + return Encoding.UTF8.GetString(applicationMessage.Payload, 0, applicationMessage.Payload.Length); + } + } +} From 7a9ddfc2b5287fb840fc08dc0ccaa099c00189d3 Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 25 Mar 2018 14:41:12 +0200 Subject: [PATCH 5/8] Fix packet identifier which is sent from the server. --- .../Adapter/MqttChannelAdapter.cs | 8 +-- .../Adapter/MqttChannelAdapterExtensions.cs | 17 ----- .../Packets/IMqttPacketWithIdentifier.cs | 2 +- .../Packets/MqttBasePublishPacket.cs | 2 +- .../Packets/MqttPacketExtensions.cs | 26 -------- .../Packets/MqttSubAckPacket.cs | 2 +- .../Packets/MqttSubscribePacket.cs | 2 +- .../Packets/MqttUnsubAckPacket.cs | 2 +- .../Packets/MqttUnsubscribe.cs | 2 +- .../Serializer/MqttPacketSerializer.cs | 65 ++++++++++++++++--- .../Server/MqttClientPendingMessagesQueue.cs | 2 +- .../Server/MqttClientSession.cs | 28 +++++--- .../Server/MqttClientSessionsManager.cs | 8 +-- .../Server/MqttClientSubscriptionsManager.cs | 13 +++- 14 files changed, 100 insertions(+), 79 deletions(-) delete mode 100644 Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapterExtensions.cs delete mode 100644 Frameworks/MQTTnet.NetStandard/Packets/MqttPacketExtensions.cs diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs index b94cf612..0ad382a1 100644 --- a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs @@ -138,7 +138,7 @@ namespace MQTTnet.Adapter } var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream(); - + var buffer = new byte[ReadBufferSize]; while (body.Length < header.BodyLength) { @@ -149,7 +149,7 @@ namespace MQTTnet.Adapter } var readBytesCount = await stream.ReadAsync(buffer, 0, bytesLeft, cancellationToken).ConfigureAwait(false); - + // Check if the client closed the connection before sending the full body. if (readBytesCount == 0) { @@ -162,7 +162,7 @@ namespace MQTTnet.Adapter } body.Seek(0L, SeekOrigin.Begin); - + return new ReceivedMqttPacket(header, body); } @@ -190,7 +190,7 @@ namespace MQTTnet.Adapter } catch (COMException comException) { - if ((uint) comException.HResult == ErrorOperationAborted) + if ((uint)comException.HResult == ErrorOperationAborted) { throw new OperationCanceledException(); } diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapterExtensions.cs b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapterExtensions.cs deleted file mode 100644 index 58258ab6..00000000 --- a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapterExtensions.cs +++ /dev/null @@ -1,17 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Packets; - -namespace MQTTnet.Adapter -{ - public static class MqttChannelAdapterExtensions - { - public static Task SendPacketsAsync(this IMqttChannelAdapter adapter, TimeSpan timeout, CancellationToken cancellationToken, params MqttBasePacket[] packets) - { - if (adapter == null) throw new ArgumentNullException(nameof(adapter)); - - return adapter.SendPacketsAsync(timeout, cancellationToken, packets); - } - } -} \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Packets/IMqttPacketWithIdentifier.cs b/Frameworks/MQTTnet.NetStandard/Packets/IMqttPacketWithIdentifier.cs index 5f7f8e97..6bbce0cb 100644 --- a/Frameworks/MQTTnet.NetStandard/Packets/IMqttPacketWithIdentifier.cs +++ b/Frameworks/MQTTnet.NetStandard/Packets/IMqttPacketWithIdentifier.cs @@ -2,6 +2,6 @@ { public interface IMqttPacketWithIdentifier { - ushort PacketIdentifier { get; set; } + ushort? PacketIdentifier { get; set; } } } diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttBasePublishPacket.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttBasePublishPacket.cs index 6218c154..ffafc537 100644 --- a/Frameworks/MQTTnet.NetStandard/Packets/MqttBasePublishPacket.cs +++ b/Frameworks/MQTTnet.NetStandard/Packets/MqttBasePublishPacket.cs @@ -2,6 +2,6 @@ { public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier { - public ushort PacketIdentifier { get; set; } + public ushort? PacketIdentifier { get; set; } } } diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttPacketExtensions.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttPacketExtensions.cs deleted file mode 100644 index a3df53c8..00000000 --- a/Frameworks/MQTTnet.NetStandard/Packets/MqttPacketExtensions.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; - -namespace MQTTnet.Packets -{ - public static class MqttPacketExtensions - { - public static TResponsePacket CreateResponse(this MqttBasePacket packet) - { - if (packet == null) throw new ArgumentNullException(nameof(packet)); - - var responsePacket = Activator.CreateInstance(); - - if (responsePacket is IMqttPacketWithIdentifier responsePacketWithIdentifier) - { - if (!(packet is IMqttPacketWithIdentifier requestPacketWithIdentifier)) - { - throw new InvalidOperationException("Response packet has PacketIdentifier but request packet does not."); - } - - responsePacketWithIdentifier.PacketIdentifier = requestPacketWithIdentifier.PacketIdentifier; - } - - return responsePacket; - } - } -} diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttSubAckPacket.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttSubAckPacket.cs index c4c1a886..3b8cf7a8 100644 --- a/Frameworks/MQTTnet.NetStandard/Packets/MqttSubAckPacket.cs +++ b/Frameworks/MQTTnet.NetStandard/Packets/MqttSubAckPacket.cs @@ -6,7 +6,7 @@ namespace MQTTnet.Packets { public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier { - public ushort PacketIdentifier { get; set; } + public ushort? PacketIdentifier { get; set; } public IList SubscribeReturnCodes { get; } = new List(); diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttSubscribePacket.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttSubscribePacket.cs index 25536e3a..63aede16 100644 --- a/Frameworks/MQTTnet.NetStandard/Packets/MqttSubscribePacket.cs +++ b/Frameworks/MQTTnet.NetStandard/Packets/MqttSubscribePacket.cs @@ -5,7 +5,7 @@ namespace MQTTnet.Packets { public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier { - public ushort PacketIdentifier { get; set; } + public ushort? PacketIdentifier { get; set; } public IList TopicFilters { get; set; } = new List(); diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubAckPacket.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubAckPacket.cs index 4797d0a4..688195fa 100644 --- a/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubAckPacket.cs +++ b/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubAckPacket.cs @@ -2,7 +2,7 @@ { public sealed class MqttUnsubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier { - public ushort PacketIdentifier { get; set; } + public ushort? PacketIdentifier { get; set; } public override string ToString() { diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs index 2c47cbbd..6c24a477 100644 --- a/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs +++ b/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs @@ -4,7 +4,7 @@ namespace MQTTnet.Packets { public sealed class MqttUnsubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier { - public ushort PacketIdentifier { get; set; } + public ushort? PacketIdentifier { get; set; } public IList TopicFilters { get; set; } = new List(); diff --git a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs index 63a3c646..0904c759 100644 --- a/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs +++ b/Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs @@ -397,7 +397,12 @@ namespace MQTTnet.Serializer private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter writer) { - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("PubRel packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); } @@ -410,7 +415,12 @@ namespace MQTTnet.Serializer if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("Publish packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); } else { @@ -444,21 +454,36 @@ namespace MQTTnet.Serializer private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer) { - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("PubAck packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck); } private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter writer) { - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("PubRec packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec); } private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter writer) { - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("PubComp packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp); } @@ -467,7 +492,12 @@ namespace MQTTnet.Serializer { if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3]."); - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("Subscribe packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); if (packet.TopicFilters?.Count > 0) { @@ -483,7 +513,12 @@ namespace MQTTnet.Serializer private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter writer) { - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("SubAck packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); if (packet.SubscribeReturnCodes?.Any() == true) { @@ -500,7 +535,12 @@ namespace MQTTnet.Serializer { if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("Unsubscribe packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); if (packet.TopicFilters?.Any() == true) { @@ -513,9 +553,14 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02); } - private static byte Serialize(IMqttPacketWithIdentifier packet, BinaryWriter writer) + private static byte Serialize(MqttUnsubAckPacket packet, BinaryWriter writer) { - writer.Write(packet.PacketIdentifier); + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("UnsubAck packet has no packet identifier."); + } + + writer.Write(packet.PacketIdentifier.Value); return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck); } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs index 27dd49d9..2469eded 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs @@ -78,7 +78,7 @@ namespace MQTTnet.Server throw new InvalidOperationException(); // should not happen } - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false); _logger.Trace("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId); } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs index aee4351c..62bb39fc 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; +using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Internal; @@ -14,6 +15,7 @@ namespace MQTTnet.Server { public sealed class MqttClientSession : IDisposable { + private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); private readonly IMqttServerOptions _options; private readonly IMqttNetLogger _logger; private readonly MqttRetainedMessagesManager _retainedMessagesManager; @@ -129,6 +131,11 @@ namespace MQTTnet.Server var publishPacket = applicationMessage.ToPublishPacket(); publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel; + if (publishPacket.QualityOfServiceLevel > 0) + { + publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); + } + PendingMessagesQueue.Enqueue(publishPacket); } @@ -205,7 +212,7 @@ namespace MQTTnet.Server if (packet is MqttPingReqPacket) { - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPingRespPacket()); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttPingRespPacket() }); } if (packet is MqttPubRelPacket pubRelPacket) @@ -215,7 +222,12 @@ namespace MQTTnet.Server if (packet is MqttPubRecPacket pubRecPacket) { - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, pubRecPacket.CreateResponse()); + var responsePacket = new MqttPubRelPacket + { + PacketIdentifier = pubRecPacket.PacketIdentifier + }; + + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { responsePacket }); } if (packet is MqttPubAckPacket || packet is MqttPubCompPacket) @@ -246,11 +258,11 @@ namespace MQTTnet.Server private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false); - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { subscribeResult.ResponsePacket }).ConfigureAwait(false); if (subscribeResult.CloseConnection) { - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttDisconnectPacket() }).ConfigureAwait(false); await StopAsync().ConfigureAwait(false); } @@ -260,7 +272,7 @@ namespace MQTTnet.Server private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false); - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { unsubscribeResult }); } private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection topicFilters) @@ -302,7 +314,7 @@ namespace MQTTnet.Server await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false); var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }; - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false); } private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -311,13 +323,13 @@ namespace MQTTnet.Server await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false); var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }; - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false); } private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }; - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index fe57552f..f1580891 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -53,10 +53,10 @@ namespace MQTTnet.Server var connectReturnCode = ValidateConnection(connectPacket); if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { - await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket + await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttConnAckPacket { ConnectReturnCode = connectReturnCode - }).ConfigureAwait(false); + }}).ConfigureAwait(false); return; } @@ -64,11 +64,11 @@ namespace MQTTnet.Server var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false); clientSession = result.Session; - await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket + await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttConnAckPacket { ConnectReturnCode = connectReturnCode, IsSessionPresent = result.IsExistingSession - }).ConfigureAwait(false); + }}).ConfigureAwait(false); ClientConnectedCallback?.Invoke(new ConnectedMqttClient { diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs index 97b9f047..e4ee9214 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs @@ -14,7 +14,7 @@ namespace MQTTnet.Server private readonly Dictionary _subscriptions = new Dictionary(); private readonly IMqttServerOptions _options; private readonly string _clientId; - + public MqttClientSubscriptionsManager(IMqttServerOptions options, string clientId) { _options = options ?? throw new ArgumentNullException(nameof(options)); @@ -30,7 +30,11 @@ namespace MQTTnet.Server var result = new MqttClientSubscribeResult { - ResponsePacket = subscribePacket.CreateResponse(), + ResponsePacket = new MqttSubAckPacket + { + PacketIdentifier = subscribePacket.PacketIdentifier + }, + CloseConnection = false }; @@ -87,7 +91,10 @@ namespace MQTTnet.Server _semaphore.Release(); } - return unsubscribePacket.CreateResponse(); + return new MqttUnsubAckPacket + { + PacketIdentifier = unsubscribePacket.PacketIdentifier + }; } public async Task CheckSubscriptionsAsync(MqttApplicationMessage applicationMessage) From 3193381b693402d564ba9d23c13ab0addbbd0f4a Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 25 Mar 2018 14:42:12 +0200 Subject: [PATCH 6/8] Refactor web socket stream. Add mqtt as default sub protocol. --- .../Client/MqttClientWebSocketOptions.cs | 2 +- .../Implementations/WebSocketStream.cs | 36 +++++++++++-------- 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientWebSocketOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientWebSocketOptions.cs index 1819daab..114283d4 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientWebSocketOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientWebSocketOptions.cs @@ -9,7 +9,7 @@ namespace MQTTnet.Client public IDictionary RequestHeaders { get; set; } - public ICollection SubProtocols { get; set; } + public ICollection SubProtocols { get; set; } = new List { "mqtt" }; public CookieContainer CookieContainer { get; set; } diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs b/Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs index 1aa02a64..bad03935 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs @@ -11,9 +11,9 @@ namespace MQTTnet.Implementations { public class WebSocketStream : Stream { - private readonly WebSocket _webSocket; - private readonly byte[] _chunkBuffer = new byte[MqttWebSocketChannel.BufferSize]; + private readonly byte[] _chunckBuffer = new byte[MqttWebSocketChannel.BufferSize]; private readonly Queue _buffer = new Queue(MqttWebSocketChannel.BufferSize); + private readonly WebSocket _webSocket; public WebSocketStream(WebSocket webSocket) { @@ -55,25 +55,28 @@ namespace MQTTnet.Implementations // Use existing date from buffer. while (count > 0 && _buffer.Any()) { - buffer[offset++] = _buffer.Dequeue(); + buffer[offset] = _buffer.Dequeue(); count--; bytesRead++; + offset++; } if (count == 0) { return bytesRead; } - + + // Fetch new data if the buffer is not full. while (_webSocket.State == WebSocketState.Open) { - await FetchChunkAsync(cancellationToken); + await FetchChunkAsync(cancellationToken).ConfigureAwait(false); while (count > 0 && _buffer.Any()) { - buffer[offset++] = _buffer.Dequeue(); + buffer[offset] = _buffer.Dequeue(); count--; bytesRead++; + offset++; } if (count == 0) @@ -111,19 +114,24 @@ namespace MQTTnet.Implementations } private async Task FetchChunkAsync(CancellationToken cancellationToken) - { - var response = await _webSocket.ReceiveAsync(new ArraySegment(_chunkBuffer, 0, _chunkBuffer.Length), cancellationToken).ConfigureAwait(false); - - for (var i = 0; i < response.Count; i++) - { - var @byte = _chunkBuffer[i]; - _buffer.Enqueue(@byte); - } + { + var response = await _webSocket.ReceiveAsync(new ArraySegment(_chunckBuffer, 0, _chunckBuffer.Length), cancellationToken).ConfigureAwait(false); if (response.MessageType == WebSocketMessageType.Close) { await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false); } + else if (response.MessageType == WebSocketMessageType.Binary) + { + for (var i = 0; i < response.Count; i++) + { + _buffer.Enqueue(_chunckBuffer[i]); + } + } + else if (response.MessageType == WebSocketMessageType.Text) + { + throw new MqttProtocolViolationException("WebSocket channel received TEXT message."); + } } } } From 3587a8ebd1651f53fac8a7cf3b2edb11b2085c27 Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 25 Mar 2018 14:42:34 +0200 Subject: [PATCH 7/8] Add new option which allows configuring the send interval for keep alive packets. --- .../Client/IMqttClientOptions.cs | 2 + .../MQTTnet.NetStandard/Client/MqttClient.cs | 51 +++++++++++++++---- .../Client/MqttClientOptions.cs | 4 +- .../Client/MqttPacketDispatcher.cs | 24 ++++++--- 4 files changed, 63 insertions(+), 18 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs index 035dba11..2f5b5042 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs @@ -13,6 +13,8 @@ namespace MQTTnet.Client TimeSpan CommunicationTimeout { get; } TimeSpan KeepAlivePeriod { get; } + TimeSpan? KeepAliveSendInterval { get; set; } + MqttProtocolVersion ProtocolVersion { get; } IMqttClientChannelOptions ChannelOptions { get; } diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs index ffdc7f26..d01a47c8 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -15,6 +16,7 @@ namespace MQTTnet.Client public class MqttClient : IMqttClient { private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); + private readonly Stopwatch _sendTracker = new Stopwatch(); private readonly IMqttClientAdapterFactory _adapterFactory; private readonly MqttPacketDispatcher _packetDispatcher; private readonly IMqttNetLogger _logger; @@ -59,10 +61,12 @@ namespace MQTTnet.Client _logger.Trace("Connection with server established."); await StartReceivingPacketsAsync(_cancellationTokenSource.Token).ConfigureAwait(false); - var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false); + var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false); _logger.Trace("MQTT connection with server established."); + _sendTracker.Restart(); + if (_options.KeepAlivePeriod != TimeSpan.Zero) { StartSendingKeepAliveMessages(_cancellationTokenSource.Token); @@ -149,7 +153,7 @@ namespace MQTTnet.Client case MqttQualityOfServiceLevel.AtMostOnce: { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosGroup).ConfigureAwait(false); + await SendAsync(qosGroup.ToArray()).ConfigureAwait(false); break; } case MqttQualityOfServiceLevel.AtLeastOnce: @@ -167,8 +171,14 @@ namespace MQTTnet.Client foreach (var publishPacket in qosGroup) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); + var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); - await SendAndReceiveAsync(pubRecPacket.CreateResponse()).ConfigureAwait(false); + var pubRelPacket = new MqttPubRelPacket + { + PacketIdentifier = pubRecPacket.PacketIdentifier + }; + + await SendAndReceiveAsync(pubRelPacket).ConfigureAwait(false); } break; @@ -325,24 +335,31 @@ namespace MQTTnet.Client private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket) { - return SendAsync(pubRelPacket.CreateResponse()); + var response = new MqttPubCompPacket + { + PacketIdentifier = pubRelPacket.PacketIdentifier + }; + + return SendAsync(response); } - private Task SendAsync(MqttBasePacket packet) + private Task SendAsync(params MqttBasePacket[] packets) { - return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packet); + _sendTracker.Restart(); + return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packets); } private async Task SendAndReceiveAsync(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket { - ushort identifier = 0; + ushort? identifier = null; if (requestPacket is IMqttPacketWithIdentifier requestPacketWithIdentifier) { identifier = requestPacketWithIdentifier.PacketIdentifier; } var packetAwaiter = _packetDispatcher.WaitForPacketAsync(typeof(TResponsePacket), identifier, _options.CommunicationTimeout); - await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false); + await SendAsync(requestPacket).ConfigureAwait(false); + return (TResponsePacket)await packetAwaiter.ConfigureAwait(false); } @@ -354,8 +371,22 @@ namespace MQTTnet.Client { while (!cancellationToken.IsCancellationRequested) { - await SendAndReceiveAsync(new MqttPingReqPacket()).ConfigureAwait(false); - await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false); + TimeSpan keepAliveSendInterval; + if (_options.KeepAliveSendInterval.HasValue) + { + keepAliveSendInterval = _options.KeepAliveSendInterval.Value; + } + else + { + keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75); + } + + if (_sendTracker.Elapsed > keepAliveSendInterval) + { + await SendAndReceiveAsync(new MqttPingReqPacket()).ConfigureAwait(false); + } + + await Task.Delay(keepAliveSendInterval, cancellationToken).ConfigureAwait(false); } } catch (OperationCanceledException) diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs index 20e711f3..9640a474 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs @@ -13,7 +13,9 @@ namespace MQTTnet.Client public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials(); - public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5); + public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15); + + public TimeSpan? KeepAliveSendInterval { get; set; } public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10); diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs index ca759311..8b09fbce 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs @@ -18,7 +18,7 @@ namespace MQTTnet.Client _logger = logger ?? throw new ArgumentNullException(nameof(logger)); } - public async Task WaitForPacketAsync(Type responseType, ushort identifier, TimeSpan timeout) + public async Task WaitForPacketAsync(Type responseType, ushort? identifier, TimeSpan timeout) { var packetAwaiter = AddPacketAwaiter(responseType, identifier); try @@ -44,13 +44,13 @@ namespace MQTTnet.Client if (_awaiters.TryGetValue(type, out var byId)) { - ushort identifier = 0; + ushort? identifier = 0; if (packet is IMqttPacketWithIdentifier packetWithIdentifier) { identifier = packetWithIdentifier.PacketIdentifier; } - if (byId.TryRemove(identifier, out var tcs)) + if (byId.TryRemove(identifier.Value, out var tcs)) { tcs.TrySetResult(packet); return; @@ -65,12 +65,17 @@ namespace MQTTnet.Client _awaiters.Clear(); } - private TaskCompletionSource AddPacketAwaiter(Type responseType, ushort identifier) + private TaskCompletionSource AddPacketAwaiter(Type responseType, ushort? identifier) { var tcs = new TaskCompletionSource(); + if (!identifier.HasValue) + { + identifier = 0; + } + var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary>()); - if (!byId.TryAdd(identifier, tcs)) + if (!byId.TryAdd(identifier.Value, tcs)) { throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{responseType}' with identifier {identifier}."); } @@ -78,10 +83,15 @@ namespace MQTTnet.Client return tcs; } - private void RemovePacketAwaiter(Type responseType, ushort identifier) + private void RemovePacketAwaiter(Type responseType, ushort? identifier) { + if (!identifier.HasValue) + { + identifier = 0; + } + var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary>()); - byId.TryRemove(identifier, out var _); + byId.TryRemove(identifier.Value, out var _); } } } \ No newline at end of file From e58cb3c9cd8472c4b2c2a3473f6a12c8f51d138b Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 25 Mar 2018 14:42:43 +0200 Subject: [PATCH 8/8] Update docs. --- Build/MQTTnet.nuspec | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 87140eb5..d3f11d56 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -2,7 +2,7 @@ MQTTnet - 2.7.1 + 2.7.2 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,15 +10,11 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). - * [Core] Fixed wrong parsing of ConnAck packet for protocol version 3.1.0. -* [Core] Log messages are now overriding ToString() and providing a ready to use text representation. -* [Client] Optimized package dispatcher and added several new exceptions. -* [Client] The _ManagedClient_ now has an event which is fired after a queued application message was processed (including exception). -* [Client] The _ManagedClient_ now supports unsubscribing (thanks to @lerppana) -* [Server] Fixed some minor async issues. -* [Server] Fixed wrong comparison of the topic and QoS for retained messages. -* [Server] Added a property which provides access to the used options (read only). -* [Server] Fixed a null ref expection when using an interceptor and publishing via the server directly. + * [Client] Added the subprotocol "mqtt" as default for web socket based connections. +* [Client] Added a new client setting called "KeepAliveSendInterval". It allows configuring the effective interval for sending ping requests. +* [Client] The client will no longer send ping requests if other packets are sent within the configured interval. +* [Server] The server now generates a valid packet identifier when disaptching publish packets to clients. +* [Core] Add several new extension methods. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin