Fix wrong QoS level handling for server.

dotnet · Aug 7, 2017 · 59b0b79 · 59b0b79
1 parent 0172606
commit 59b0b79
Show file tree

Hide file tree

Showing 15 changed files with 31 additions and 32 deletions.
diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec
@@ -2,19 +2,15 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.1.5</version>
+    <version>2.1.5.1</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
     <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
-    <releaseNotes>* [MqttServer] Added support for publishing application messages
-* [Core] Fixed QoS level 2 handling
-* [Core] Performance optimizations
-* [MqttClient/MqttServer] Errors while handline application messages are now catched and traced
-* [MqttClient/MqttServer] Added interfaces
+    <releaseNotes>* [Server] Fixed wrong handling of QoS levels
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2017</copyright>
     <tags>MQTT MQTTClient MQTTServer MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Queue Hardware</tags> 

diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs
@@ -86,7 +86,7 @@ private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cance
                 try
                 {
                     var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null);
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception) when (!(exception is ObjectDisposedException))
@@ -107,7 +107,7 @@ private async Task AcceptTlsEndpointConnectionsAsync(CancellationToken cancellat
                     var sslStream = new SslStream(new NetworkStream(clientSocket));
                     await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false);
 
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception)

diff --git a/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs b/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs
@@ -12,7 +12,7 @@ public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new DefaultMqttV311PacketSerializer()));
+            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new MqttV311PacketSerializer()));
         }
     }
 }
diff --git a/Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs b/Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs
@@ -11,5 +11,5 @@
 [assembly: AssemblyCulture("")]
 [assembly: ComVisible(false)]
 [assembly: Guid("a480ef90-0eaa-4d9a-b271-47a9c47f6f7d")]
-[assembly: AssemblyVersion("2.1.4.0")]
-[assembly: AssemblyFileVersion("2.1.4.0")]
+[assembly: AssemblyVersion("2.1.5.1")]
+[assembly: AssemblyFileVersion("2.1.5.1")]
diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs
@@ -84,7 +84,7 @@ private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cance
                 try
                 {
                     var clientSocket = await _defaultEndpointSocket.AcceptAsync();
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception)
@@ -105,7 +105,7 @@ private async Task AcceptTlsEndpointConnectionsAsync(CancellationToken cancellat
                     var sslStream = new SslStream(new NetworkStream(clientSocket));
                     await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false);
 
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception)

diff --git a/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj b/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj
@@ -4,8 +4,8 @@
     <TargetFramework>netstandard1.3</TargetFramework>
     <AssemblyName>MQTTnet</AssemblyName>
     <RootNamespace>MQTTnet</RootNamespace>
-    <AssemblyVersion>2.1.4.0</AssemblyVersion>
-    <FileVersion>2.1.4.0</FileVersion>
+    <AssemblyVersion>2.1.5.1</AssemblyVersion>
+    <FileVersion>2.1.5.1</FileVersion>
     <Version>0.0.0.0</Version>
     <Company />
     <Product />

diff --git a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs
@@ -12,7 +12,7 @@ public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new DefaultMqttV311PacketSerializer()));
+            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new MqttV311PacketSerializer()));
         }
     }
 }
diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs
@@ -52,7 +52,7 @@ private void AcceptDefaultEndpointConnectionsAsync(StreamSocketListener sender,
         {
             try
             {
-                var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(args.Socket), new DefaultMqttV311PacketSerializer());
+                var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(args.Socket), new MqttV311PacketSerializer());
                 ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(args.Socket.Information.RemoteAddress.ToString(), clientAdapter));
             }
             catch (Exception exception)

diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs
@@ -12,7 +12,7 @@ public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new DefaultMqttV311PacketSerializer()));
+            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new MqttV311PacketSerializer()));
         }
     }
 }
diff --git a/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs b/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs
@@ -10,5 +10,5 @@
 [assembly: AssemblyTrademark("")]
 [assembly: AssemblyCulture("")]
 [assembly: ComVisible(false)]
-[assembly: AssemblyVersion("2.1.4.0")]
-[assembly: AssemblyFileVersion("2.1.4.0")]
+[assembly: AssemblyVersion("2.1.5.1")]
+[assembly: AssemblyFileVersion("2.1.5.1")]
diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs
@@ -267,7 +267,7 @@ private Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
                 return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
             }
 
-            throw new InvalidOperationException();
+            throw new MqttCommunicationException("Received a not supported QoS level.");
         }
 
         private async Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)

diff --git a/MQTTnet.Core/MQTTnet.Core.csproj b/MQTTnet.Core/MQTTnet.Core.csproj
@@ -16,8 +16,8 @@
     <PackageIconUrl></PackageIconUrl>
     <RepositoryUrl></RepositoryUrl>
     <PackageTags></PackageTags>
-    <FileVersion>2.1.4.0</FileVersion>
-    <AssemblyVersion>2.1.4.0</AssemblyVersion>
+    <FileVersion>2.1.5.1</FileVersion>
+    <AssemblyVersion>2.1.5.1</AssemblyVersion>
     <PackageLicenseUrl></PackageLicenseUrl>
   </PropertyGroup>
 

diff --git a/...alizer/DefaultMqttV311PacketSerializer.cs → ...re/Serializer/MqttV311PacketSerializer.cs b/...alizer/DefaultMqttV311PacketSerializer.cs → ...re/Serializer/MqttV311PacketSerializer.cs
@@ -9,8 +9,10 @@
 
 namespace MQTTnet.Core.Serializer
 {
-    public sealed class DefaultMqttV311PacketSerializer : IMqttPacketSerializer
+    public sealed class MqttV311PacketSerializer : IMqttPacketSerializer
     {
+        private static readonly byte[] MqttV311Prefix = Encoding.UTF8.GetBytes("MQTT");
+
         public Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
@@ -351,8 +353,6 @@ private static void ValidatePublishPacket(MqttPublishPacket packet)
             }
         }
 
-        private static readonly byte[] MqttPrefix = Encoding.UTF8.GetBytes("MQTT");
-
         private static Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
         {
             ValidateConnectPacket(packet);
@@ -361,7 +361,7 @@ private static Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationC
             {
                 // Write variable header
                 output.Write(0x00, 0x04); // 3.1.2.1 Protocol Name
-                output.Write(MqttPrefix);
+                output.Write(MqttV311Prefix);
                 output.Write(0x04); // 3.1.2.2 Protocol Level
 
                 var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags

diff --git a/MQTTnet.Core/Server/MqttClientSession.cs b/MQTTnet.Core/Server/MqttClientSession.cs
@@ -153,13 +153,16 @@ private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
                 _publishPacketReceivedCallback(this, publishPacket);
+                return Task.FromResult(0);
             }
-            else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+
+            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
                 _publishPacketReceivedCallback(this, publishPacket);
                 return _adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
-            else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+
+            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
             {
                 // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
                 lock (_unacknowledgedPublishPackets)
@@ -172,7 +175,7 @@ private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
                 return _adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
 
-            throw new MqttCommunicationException("Received not supported QoS level.");
+            throw new MqttCommunicationException("Received a not supported QoS level.");
         }
 
         private Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket)

diff --git a/Tests/MQTTnet.Core.Tests/DefaultMqttV311PacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/DefaultMqttV311PacketSerializerTests.cs
@@ -405,7 +405,7 @@ public byte[] ToArray()
 
         private void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
         {
-            var serializer = new DefaultMqttV311PacketSerializer();
+            var serializer = new MqttV311PacketSerializer();
             var channel = new TestChannel();
             serializer.SerializeAsync(packet, channel).Wait();
             var buffer = channel.ToArray();
@@ -415,7 +415,7 @@ private void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Val
 
         private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
         {
-            var serializer = new DefaultMqttV311PacketSerializer();
+            var serializer = new MqttV311PacketSerializer();
 
             var channel1 = new TestChannel();
             serializer.SerializeAsync(packet, channel1).Wait();