diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index 9ebc7408..976ad12a 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/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)); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs index 9a52a98d..a34aa536 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs +++ b/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) { diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index d501c8eb..cd304956 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/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); } } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 5800bf43..533d4f0b 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/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 ConnectClientAsync(Action 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) diff --git a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs index 16405de2..893bc9eb 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs +++ b/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); + } + } } }