Fixed bug in PubRel packet generation (MQTTv5 only).

Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs

@@ -10,15 +10,9 @@ namespace MQTTnet.Formatter.V5
     {
         private readonly IMqttPacketWriter _packetWriter;
 
-        public MqttV500PacketEncoder()
-            : this(new MqttPacketWriter())
-        {
-
-        }
-
         public MqttV500PacketEncoder(IMqttPacketWriter packetWriter)
         {
-            _packetWriter = packetWriter;
+            _packetWriter = packetWriter ?? throw new ArgumentNullException(nameof(packetWriter));
         }
 
 

Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs

@@ -4,15 +4,10 @@ using MQTTnet.Packets;
 
 namespace MQTTnet.Formatter.V5
 {
-    public class MqttV500PacketFormatter : IMqttPacketFormatter
+    public sealed class MqttV500PacketFormatter : IMqttPacketFormatter
     {
-        private readonly MqttV500PacketEncoder _encoder;
-        private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder();
-
-        public MqttV500PacketFormatter()
-        {
-            _encoder = new MqttV500PacketEncoder();
-        }
+        readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder();
+        readonly MqttV500PacketEncoder _encoder;
 
         public MqttV500PacketFormatter(IMqttPacketWriter writer)
         {

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -452,7 +452,12 @@ namespace MQTTnet.Server
                             await SendAsync(publishPacket).ConfigureAwait(false);
                             await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
 
-                            await SendAsync(new MqttPubRelPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false);
+                            await SendAsync(new MqttPubRelPacket
+                            {
+                                PacketIdentifier = publishPacket.PacketIdentifier,
+                                ReasonCode = MqttPubRelReasonCode.Success
+                            }, cancellationToken).ConfigureAwait(false);
+
                             await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
                         }
                     }

Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs

@@ -5,6 +5,7 @@ using MQTTnet.Diagnostics;
 using MQTTnet.Server;
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
 
@@ -43,6 +44,8 @@ namespace MQTTnet.Tests.Mockups
 
             ServerLogger.LogMessagePublished += (s, e) =>
             {
+                Debug.WriteLine(e.LogMessage.ToString());
+
                 if (e.LogMessage.Level == MqttNetLogLevel.Error)
                 {
                     lock (_serverErrors)
@@ -54,6 +57,8 @@ namespace MQTTnet.Tests.Mockups
 
             ClientLogger.LogMessagePublished += (s, e) =>
             {
+                Debug.WriteLine(e.LogMessage.ToString());
+
                 if (e.LogMessage.Level == MqttNetLogLevel.Error)
                 {
                     lock (_clientErrors)
@@ -125,6 +130,18 @@ namespace MQTTnet.Tests.Mockups
             return client;
         }
 
+        public async Task<IMqttClient> ConnectClientAsync(Action<MqttClientOptionsBuilder> optionsBuilder)
+        {
+            var options = new MqttClientOptionsBuilder();
+            options = options.WithTcpServer("localhost", ServerPort);
+            optionsBuilder?.Invoke(options);
+
+            var client = CreateClient();
+            await client.ConnectAsync(options.Build()).ConfigureAwait(false);
+
+            return client;
+        }
+
         public void ThrowIfLogErrors()
         {
             lock (_serverErrors)

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -15,6 +15,10 @@ using System.Net.Sockets;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Client.Receiving;
+using MQTTnet.Diagnostics;
+using MQTTnet.Extensions.ManagedClient;
+using MQTTnet.Formatter;
 
 namespace MQTTnet.Tests
 {
@@ -741,5 +745,41 @@ namespace MQTTnet.Tests
                 Assert.AreEqual(null, receivedMessage.Payload);
             }
         }
+
+        [TestMethod]
+        public async Task Subscribe_With_QoS2()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+                var client1 = await testEnvironment.ConnectClientAsync(o => o.WithProtocolVersion(MqttProtocolVersion.V500));
+                var client2 = await testEnvironment.ConnectClientAsync(o => o.WithProtocolVersion(MqttProtocolVersion.V500));
+
+                var disconnectedFired = false;
+                client1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(c =>
+                {
+                    disconnectedFired = true;
+                });
+
+                var messageReceived = false;
+                client1.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c =>
+                {
+                    messageReceived = true;
+                });
+
+                await client1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("topic1").WithExactlyOnceQoS().Build());
+
+                await Task.Delay(500);
+
+                var message = new MqttApplicationMessageBuilder().WithTopic("topic1").WithPayload("Hello World").WithExactlyOnceQoS().WithRetainFlag().Build();
+                
+                await client2.PublishAsync(message);
+                await Task.Delay(500);
+
+                Assert.IsTrue(messageReceived);
+                Assert.IsTrue(client1.IsConnected);
+                Assert.IsFalse(disconnectedFired);
+            }
+        }
     }
 }