YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-16 17:23:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

1546 possible bug in server repeatedly sending the same message to client subscribed with cleansession=false (#1569)

Browse Source
This commit is contained in:
Christian
2022-11-11 08:14:15 +01:00
committed by GitHub
parent bfec9dc5ad
commit db1fc7625d
18 changed files with 475 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ReleaseNotes.md
+2 -1
View File
@@ -1,7 +1,8 @@
* [Core] Fixed several dead locks by implementing a new async signal (#1552)
* [Client] All waiting tasks will now fail when the client connection gets closed. This fixes several freezes (#1561)
* [ManagedClient] The topic is no longer declared as invalid if a topic alias is used (#1395).
* [ManagedClient] The topic is no longer declared as invalid if a topic alias is used (#1395)
* [Server] Fixed duplicated invocation of the event _ClientAcknowledgedPublishPacketAsync_ for QoS level 2 (#1550)
* [Server] Fixed issue in upgrading and downgrading of QoS levels for subscriptions and retained messages (#1560)
* [Server] Fixed memory leak when old sessions are discarded (#1553)
* [Server] When using MQTTv5 the reason code _NoMatchingSubscribers_ is returned when no client is subscribed to a published topic (#802)
* [Server] Fix sending of duplicate messages with QoS > 0 when session is recovered and session is reused (#1546)
Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs
+1 -1
View File
@@ -43,7 +43,7 @@ namespace MQTTnet.Extensions.ManagedClient
Task StartAsync(ManagedMqttClientOptions options);
Task StopAsync();
Task StopAsync(bool cleanDisconnect = true);
Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters);
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+9 -3
View File
@@ -62,6 +62,7 @@ namespace MQTTnet.Extensions.ManagedClient
CancellationTokenSource _publishingCancellationToken;
ManagedMqttClientStorageManager _storageManager;
bool _isCleanDisconnect;
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
{
@@ -255,10 +256,12 @@ namespace MQTTnet.Extensions.ManagedClient
_logger.Info("Started");
}
public async Task StopAsync()
public async Task StopAsync(bool cleanDisconnect = true)
{
ThrowIfDisposed();
_isCleanDisconnect = cleanDisconnect;
StopPublishing();
StopMaintainingConnection();
@@ -382,9 +385,12 @@ namespace MQTTnet.Extensions.ManagedClient
{
try
{
using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
if (_isCleanDisconnect)
{
await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
{
await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
}
}
}
catch (OperationCanceledException)
Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs
+7 -3
View File
@@ -64,7 +64,7 @@ namespace MQTTnet.Tests.Client
await client.ConnectAsync(options, CancellationToken.None).ConfigureAwait(false);
}
[TestMethod]
[ExpectedException(typeof(MqttCommunicationException))]
public async Task Connect_To_Wrong_Host()
@@ -80,6 +80,8 @@ namespace MQTTnet.Tests.Client
{
using (var testEnvironment = CreateTestEnvironment())
{
testEnvironment.IgnoreServerLogErrors = true;
testEnvironment.ServerPort = 8364;
var server = await testEnvironment.StartServer();
@@ -116,6 +118,8 @@ namespace MQTTnet.Tests.Client
{
using (var testEnvironment = CreateTestEnvironment())
{
testEnvironment.IgnoreServerLogErrors = true;
var server = await testEnvironment.StartServer();
using (var lowLevelClient = testEnvironment.CreateLowLevelClient())
@@ -123,7 +127,7 @@ namespace MQTTnet.Tests.Client
Assert.IsFalse(lowLevelClient.IsConnected);
var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithTimeout(TimeSpan.FromSeconds(1)).Build();
await lowLevelClient.ConnectAsync(clientOptions, CancellationToken.None);
Assert.IsTrue(lowLevelClient.IsConnected);
@@ -148,7 +152,7 @@ namespace MQTTnet.Tests.Client
catch
{
}
Assert.IsFalse(lowLevelClient.IsConnected);
}
}
Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs
+60 -1
View File
@@ -11,6 +11,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
@@ -21,6 +22,65 @@ namespace MQTTnet.Tests.Client
[TestClass]
public sealed class ManagedMqttClient_Tests : BaseTestClass
{
[TestMethod]
public async Task Receive_While_Not_Cleanly_Disconnected()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
await testEnvironment.StartServer(o => o.WithPersistentSessions());
var senderClient = await testEnvironment.ConnectClient();
// Prepare managed client.
var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
await managedClient.SubscribeAsync("#");
var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
var managedClientOptions = new ManagedMqttClientOptions
{
ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
.WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
.WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
.WithCleanSession(false)
.Build()
};
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
// Send test data.
await senderClient.PublishStringAsync("topic1");
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(1);
// Stop the managed client but keep session at server (not clean disconnect required).
await managedClient.StopAsync(false);
await LongTestDelay();
// Send new messages in the meantime.
await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
await LongTestDelay();
// Start the managed client, it should receive the new message.
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(2);
// Stop and start again, no new message should be received.
for (var i = 0; i < 3; i++)
{
await managedClient.StopAsync(false);
await LongTestDelay();
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
}
receivedMessages.AssertReceivedCountEquals(2);
}
}
[TestMethod]
public async Task Connect_To_Invalid_Server()
{
@@ -33,7 +93,6 @@ namespace MQTTnet.Tests.Client
var managedClient = testEnvironment.Factory.CreateManagedMqttClient(mqttClient);
ConnectingFailedEventArgs connectingFailedEventArgs = null;
managedClient.ConnectingFailedAsync += args =>
{
connectingFailedEventArgs = args;
Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs
+43
View File
@@ -0,0 +1,43 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Internal;
using MQTTnet.Packets;
namespace MQTTnet.Tests.Internal
{
[TestClass]
public sealed class MqttPacketBusItem_Tests
{
[TestMethod]
public void Fire_Completed_Event()
{
var eventFired = false;
var item = new MqttPacketBusItem(new MqttPublishPacket());
item.Completed += (_, __) =>
{
eventFired = true;
};
item.Complete();
Assert.IsTrue(eventFired);
}
[TestMethod]
[ExpectedException(typeof(TaskCanceledException))]
public async Task Wait_Packet_Bus_Item_After_Already_Canceled()
{
var item = new MqttPacketBusItem(new MqttPublishPacket());
// Finish the item before the actual
item.Cancel();
await item.WaitAsync();
}
}
}
Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs
+84 -84
View File
@@ -13,69 +13,26 @@ namespace MQTTnet.Tests.Internal
{
[TestClass]
public sealed class MqttPacketBus_Tests
{
[TestMethod]
[ExpectedException(typeof(TaskCanceledException))]
public async Task Wait_With_Empty_Bus()
{
var bus = new MqttPacketBus();
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
[TestMethod]
public async Task Fill_From_Different_Task()
{
const int MessageCount = 500;
var delayRandom = new Random();
var bus = new MqttPacketBus();
_ = Task.Run(
() =>
{
for (int i = 0; i < MessageCount; i++)
{
bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health);
Thread.Sleep(delayRandom.Next(0, 10));
}
});
for (var i = 0; i < MessageCount; i++)
{
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
Assert.AreEqual(0, bus.TotalItemsCount);
}
{
[TestMethod]
public void Alternate_Priorities()
{
var bus = new MqttPacketBus();
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
Assert.AreEqual(9, bus.TotalItemsCount);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
@@ -85,33 +42,10 @@ namespace MQTTnet.Tests.Internal
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
Assert.AreEqual(0, bus.TotalItemsCount);
}
[TestMethod]
public void Export_Packets_Without_Dequeue()
{
var bus = new MqttPacketBus();
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
Assert.AreEqual(3, bus.TotalItemsCount);
var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
Assert.AreEqual(3, exportedPackets.Count);
Assert.AreEqual(3, bus.TotalItemsCount);
}
[TestMethod]
public void Await_Single_Packet()
{
@@ -121,31 +55,97 @@ namespace MQTTnet.Tests.Internal
var item1 = new MqttPacketBusItem(new MqttPublishPacket());
var item2 = new MqttPacketBusItem(new MqttPublishPacket());
var item3 = new MqttPacketBusItem(new MqttPublishPacket());
item3.Delivered += (_, __) =>
item3.Completed += (_, __) =>
{
delivered = true;
};
bus.EnqueueItem(item1, MqttPacketBusPartition.Data);
bus.EnqueueItem(item2, MqttPacketBusPartition.Data);
bus.EnqueueItem(item3, MqttPacketBusPartition.Data);
Assert.IsFalse(delivered);
bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
Assert.IsFalse(delivered);
bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
Assert.IsFalse(delivered);
bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
// The third packet has the event attached.
Assert.IsTrue(delivered);
}
[TestMethod]
public void Export_Packets_Without_Dequeue()
{
var bus = new MqttPacketBus();
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
Assert.AreEqual(3, bus.TotalItemsCount);
var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
Assert.AreEqual(3, exportedPackets.Count);
Assert.AreEqual(3, bus.TotalItemsCount);
}
[TestMethod]
public async Task Fill_From_Different_Task()
{
const int MessageCount = 500;
var delayRandom = new Random();
var bus = new MqttPacketBus();
_ = Task.Run(
() =>
{
for (var i = 0; i < MessageCount; i++)
{
bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health);
Thread.Sleep(delayRandom.Next(0, 10));
}
});
for (var i = 0; i < MessageCount; i++)
{
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
Assert.AreEqual(0, bus.TotalItemsCount);
}
[TestMethod]
[ExpectedException(typeof(OperationCanceledException))]
public async Task Wait_With_Empty_Bus()
{
var bus = new MqttPacketBus();
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
}
}
Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs
+27 -2
View File
@@ -2,8 +2,10 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
@@ -17,7 +19,12 @@ namespace MQTTnet.Tests.Mockups
public TestApplicationMessageReceivedHandler(IMqttClient mqttClient)
{
mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync;
if (mqttClient == null)
{
throw new ArgumentNullException(nameof(mqttClient));
}
mqttClient.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;
}
public List<MqttApplicationMessageReceivedEventArgs> ReceivedEventArgs
@@ -39,7 +46,25 @@ namespace MQTTnet.Tests.Mockups
}
}
Task MqttClientOnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
public string GeneratePayloadSequence()
{
var sequence = new StringBuilder();
lock (_receivedEventArgs)
{
foreach (var receivedEventArg in _receivedEventArgs)
{
var payload = receivedEventArg.ApplicationMessage.ConvertPayloadToString();
// An empty payload is not part of the sequence!
sequence.Append(payload);
}
}
return sequence.ToString();
}
Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
{
lock (_receivedEventArgs)
{
Source/MQTTnet.Tests/Mockups/TestEnvironment.cs
+16 -5
View File
@@ -11,6 +11,7 @@ using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Extensions.Rpc;
using MQTTnet.Formatter;
using MQTTnet.Internal;
@@ -109,7 +110,7 @@ namespace MQTTnet.Tests.Mockups
configureOptions.Invoke(optionsBuilder);
var options = optionsBuilder.Build();
var client = CreateClient();
if (timeout == TimeSpan.Zero)
@@ -137,7 +138,7 @@ namespace MQTTnet.Tests.Mockups
options = options.WithTcpServer("127.0.0.1", ServerPort);
var client = CreateClient();
if (timeout == TimeSpan.Zero)
{
await client.ConnectAsync(options.Build()).ConfigureAwait(false);
@@ -161,7 +162,7 @@ namespace MQTTnet.Tests.Mockups
}
var client = CreateClient();
if (timeout == TimeSpan.Zero)
{
await client.ConnectAsync(options).ConfigureAwait(false);
@@ -199,6 +200,16 @@ namespace MQTTnet.Tests.Mockups
return new TestApplicationMessageReceivedHandler(mqttClient);
}
public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(IManagedMqttClient managedClient)
{
if (managedClient == null)
{
throw new ArgumentNullException(nameof(managedClient));
}
return new TestApplicationMessageReceivedHandler(managedClient.InternalClient);
}
public IMqttClient CreateClient()
{
lock (_clients)
@@ -325,7 +336,7 @@ namespace MQTTnet.Tests.Mockups
var options = optionsBuilder.Build();
var server = CreateServer(options);
await server.StartAsync();
// The OS has chosen the port to we have to properly expose it to the tests.
ServerPort = options.DefaultEndpointOptions.Port;
return server;
@@ -344,7 +355,7 @@ namespace MQTTnet.Tests.Mockups
var options = optionsBuilder.Build();
var server = CreateServer(options);
await server.StartAsync();
// The OS has chosen the port to we have to properly expose it to the tests.
ServerPort = options.DefaultEndpointOptions.Port;
return server;
Source/MQTTnet.Tests/Server/QoS_Tests.cs
+51
View File
@@ -15,6 +15,57 @@ namespace MQTTnet.Tests.Server
[TestClass]
public sealed class QoS_Tests : BaseTestClass
{
[TestMethod]
public async Task Preserve_Message_Order_For_Queued_Messages()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer(o => o.WithPersistentSessions());
// Create a session which will contain the messages.
var dummyClient = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
await dummyClient.SubscribeAsync("#", MqttQualityOfServiceLevel.AtLeastOnce);
dummyClient.Dispose();
// Now inject messages which are appended to the queue of the client.
await server.InjectApplicationMessage("T", "0", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "2", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "1", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "4", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "3", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "6", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "5", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "8", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "7", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "9", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await LongTestDelay();
// Create a new client for the existing message.
var client = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
var messages = testEnvironment.CreateApplicationMessageHandler(client);
await LongTestDelay();
var payloadSequence = messages.GeneratePayloadSequence();
Assert.AreEqual("0214365879", payloadSequence);
// Disconnect and reconnect to make sure that the server will not send the messages twice.
await client.DisconnectAsync();
await LongTestDelay();
await client.ReconnectAsync();
await LongTestDelay();
payloadSequence = messages.GeneratePayloadSequence();
Assert.AreEqual("0214365879", payloadSequence);
}
}
[TestMethod]
public async Task Fire_Event_On_Client_Acknowledges_QoS_0()
{
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+2 -2
View File
@@ -86,7 +86,7 @@ namespace MQTTnet.Adapter
await Task.WhenAny(connectTask, timeout.Task).ConfigureAwait(false);
if (timeout.Task.IsCompleted && !connectTask.IsCompleted)
{
throw new OperationCanceledException("MQTT connect cancelled.", cancellationToken);
throw new OperationCanceledException("MQTT connect canceled.", cancellationToken);
}
// Make sure that the exception from the connect task gets thrown.
@@ -317,7 +317,7 @@ namespace MQTTnet.Adapter
if (cancellationToken.IsCancellationRequested)
{
return ReadFixedHeaderResult.Cancelled;
return ReadFixedHeaderResult.Canceled;
}
if (bytesRead == 0)
Source/MQTTnet/Client/MqttClientExtensions.cs
+31 -20
View File
@@ -14,26 +14,6 @@ namespace MQTTnet.Client
{
public static class MqttClientExtensions
{
public static async Task<bool> TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default)
{
if (client == null)
{
throw new ArgumentNullException(nameof(client));
}
try
{
await client.PingAsync(cancellationToken).ConfigureAwait(false);
return true;
}
catch
{
// Ignore errors.
}
return false;
}
public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null)
{
if (client == null)
@@ -89,6 +69,17 @@ namespace MQTTnet.Client
return mqttClient.PublishBinaryAsync(topic, payloadBuffer, qualityOfServiceLevel, retain, cancellationToken);
}
public static Task ReconnectAsync(this IMqttClient client, CancellationToken cancellationToken = default)
{
if (client.Options == null)
{
throw new InvalidOperationException(
"The MQTT client was not connected before. A reconnect is only permitted when the client was already connected or at least tried to.");
}
return client.ConnectAsync(client.Options, cancellationToken);
}
public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data)
{
if (client == null)
@@ -137,6 +128,26 @@ namespace MQTTnet.Client
return mqttClient.SubscribeAsync(subscribeOptions, cancellationToken);
}
public static async Task<bool> TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default)
{
if (client == null)
{
throw new ArgumentNullException(nameof(client));
}
try
{
await client.PingAsync(cancellationToken).ConfigureAwait(false);
return true;
}
catch
{
// Ignore errors.
}
return false;
}
public static Task<MqttClientUnsubscribeResult> UnsubscribeAsync(this IMqttClient mqttClient, string topic, CancellationToken cancellationToken = default)
{
if (mqttClient == null)
Source/MQTTnet/Formatter/ReadFixedHeaderResult.cs
+3 -3
View File
@@ -6,9 +6,9 @@ namespace MQTTnet.Formatter
{
public struct ReadFixedHeaderResult
{
public static ReadFixedHeaderResult Cancelled { get; } = new ReadFixedHeaderResult
public static ReadFixedHeaderResult Canceled { get; } = new ReadFixedHeaderResult
{
IsCancelled = true
IsCanceled = true
};
public static ReadFixedHeaderResult ConnectionClosed { get; } = new ReadFixedHeaderResult
@@ -16,7 +16,7 @@ namespace MQTTnet.Formatter
IsConnectionClosed = true
};
public bool IsCancelled { get; set; }
public bool IsCanceled { get; set; }
public bool IsConnectionClosed { get; set; }
Source/MQTTnet/Internal/MqttPacketBusItem.cs
+22 -26
View File
@@ -10,40 +10,36 @@ namespace MQTTnet.Internal
{
public sealed class MqttPacketBusItem
{
readonly AsyncTaskCompletionSource<int> _promise = new AsyncTaskCompletionSource<int>();
readonly AsyncTaskCompletionSource<bool> _promise = new AsyncTaskCompletionSource<bool>();
public MqttPacketBusItem(MqttPacket packet)
{
Packet = packet ?? throw new ArgumentNullException(nameof(packet));
}
public event EventHandler Completed;
public MqttPacket Packet { get; }
public event EventHandler Delivered;
public Task WaitForDeliveryAsync()
{
return _promise.Task;
}
public void MarkAsDelivered()
{
if (_promise.TrySetResult(0))
{
Delivered?.Invoke(this, EventArgs.Empty);
}
}
public void MarkAsFailed(Exception exception)
{
if (exception == null) throw new ArgumentNullException(nameof(exception));
_promise.TrySetException(exception);
}
public void MarkAsCancelled()
public void Cancel()
{
_promise.TrySetCanceled();
}
public void Complete()
{
_promise.TrySetResult(true);
Completed?.Invoke(this, EventArgs.Empty);
}
public void Fail(Exception exception)
{
_promise.TrySetException(exception);
}
public Task WaitAsync()
{
return _promise.Task;
}
}
}
Source/MQTTnet/Server/Internal/MqttClient.cs
+4 -4
View File
@@ -507,15 +507,15 @@ namespace MQTTnet.Server
try
{
await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);
packetBusItem.MarkAsDelivered();
packetBusItem.Complete();
}
catch (OperationCanceledException)
{
packetBusItem.MarkAsCancelled();
packetBusItem.Cancel();
}
catch (Exception exception)
{
packetBusItem.MarkAsFailed(exception);
packetBusItem.Fail(exception);
}
finally
{
@@ -563,7 +563,7 @@ namespace MQTTnet.Server
catch (ObjectDisposedException)
{
// This can happen when connections are created and dropped very quickly.
// It is not an issue if the cancellation token cannot be cancelled multiple times.
// It is not an issue if the cancellation token cannot be canceled multiple times.
}
}
Source/MQTTnet/Server/Internal/MqttSession.cs
+43 -7
View File
@@ -4,7 +4,6 @@
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@@ -25,7 +24,8 @@ namespace MQTTnet.Server
readonly MqttServerOptions _serverOptions;
readonly MqttClientSubscriptionsManager _subscriptionsManager;
readonly ConcurrentDictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
// Do not use a dictionary in order to keep the ordering of the messages.
readonly List<MqttPublishPacket> _unacknowledgedPublishPackets = new List<MqttPublishPacket>();
// Bookkeeping to know if this is a subscribing client; lazy initialize later.
HashSet<string> _subscribedTopics;
@@ -72,7 +72,14 @@ namespace MQTTnet.Server
public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier)
{
_unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket);
MqttPublishPacket publishPacket;
lock (_unacknowledgedPublishPackets)
{
publishPacket = _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier));
_unacknowledgedPublishPackets.Remove(publishPacket);
}
return publishPacket;
}
@@ -130,7 +137,10 @@ namespace MQTTnet.Server
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
_unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
lock (_unacknowledgedPublishPackets)
{
_unacknowledgedPublishPackets.Add(publishPacket);
}
}
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
@@ -145,8 +155,10 @@ namespace MQTTnet.Server
{
// This will only return the matching PUBLISH packet but does not remove it.
// This is required for QoS 2.
_unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket);
return publishPacket;
lock (_unacknowledgedPublishPackets)
{
return _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier));
}
}
public void Recover()
@@ -154,9 +166,33 @@ namespace MQTTnet.Server
// TODO: Keep the bus and only insert pending items again.
// TODO: Check if packet identifier must be restarted or not.
// TODO: Recover package identifier.
/*
The Session state in the Client consists of:
· QoS 1 and QoS 2 messages which have been sent to the Server, but have not been completely acknowledged.
· QoS 2 messages which have been received from the Server, but have not been completely acknowledged.
The Session state in the Server consists of:
· The existence of a Session, even if the rest of the Session state is empty.
· The Client’s subscriptions.
· QoS 1 and QoS 2 messages which have been sent to the Client, but have not been completely acknowledged.
· QoS 1 and QoS 2 messages pending transmission to the Client.
· QoS 2 messages which have been received from the Client, but have not been completely acknowledged.
· Optionally, QoS 0 messages pending transmission to the Client.
*/
// Create a copy of all currently unacknowledged publish packets and clear the storage.
// We must re-enqueue them in order to trigger other code.
List<MqttPublishPacket> unacknowledgedPublishPackets;
lock (_unacknowledgedPublishPackets)
{
unacknowledgedPublishPackets = _unacknowledgedPublishPackets.ToList();
_unacknowledgedPublishPackets.Clear();
}
_packetBus.Clear();
foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList())
foreach (var publishPacket in unacknowledgedPublishPackets)
{
EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
Source/MQTTnet/Server/MqttServerExtensions.cs
+69 -9
View File
@@ -3,29 +3,89 @@
// See the LICENSE file in the project root for more information.
using System;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public static class MqttServerExtensions
{
public static Task InjectApplicationMessage(
this MqttServer server,
string topic,
string payload = null,
MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
bool retain = false)
{
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
if (topic == null)
{
throw new ArgumentNullException(nameof(topic));
}
var payloadBuffer = EmptyBuffer.Array;
if (payload is string stringPayload)
{
payloadBuffer = Encoding.UTF8.GetBytes(stringPayload);
}
return server.InjectApplicationMessage(
new InjectedMqttApplicationMessage(
new MqttApplicationMessage
{
Topic = topic,
Payload = payloadBuffer,
QualityOfServiceLevel = qualityOfServiceLevel,
Retain = retain
}));
}
public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
if (clientId == null)
{
throw new ArgumentNullException(nameof(clientId));
}
if (topicFilters == null)
{
throw new ArgumentNullException(nameof(topicFilters));
}
return server.SubscribeAsync(clientId, topicFilters);
}
public static Task SubscribeAsync(this MqttServer server, string clientId, string topic)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topic == null) throw new ArgumentNullException(nameof(topic));
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build());
if (clientId == null)
{
throw new ArgumentNullException(nameof(clientId));
}
if (topic == null)
{
throw new ArgumentNullException(nameof(topic));
}
var topicFilters = new MqttTopicFilterBuilder().WithTopic(topic).Build();
return server.SubscribeAsync(clientId, topicFilters);
}
}
}
}
Source/MQTTnet/Server/Status/MqttSessionStatus.cs
+1 -1
View File
@@ -45,7 +45,7 @@ namespace MQTTnet.Server
var packetBusItem = new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage));
_session.EnqueueDataPacket(packetBusItem);
return packetBusItem.WaitForDeliveryAsync();
return packetBusItem.WaitAsync();
}
public Task DeleteAsync()