Compare commits

...
Author SHA1 Message Date
christian f6357c42b6 Prepare recommended fix 2026-06-16 19:10:11 +02:00
christian 16512d5b89 Create Unit Test 2026-06-16 19:09:58 +02:00
christian 745d1d25fe Update release notes 2026-06-13 18:07:57 +02:00
christian 905f070c37 Reset release notes 2026-06-13 18:06:14 +02:00
Suhas Holla Karkada Chandrashekar bca2ded8eb Fix MQTT v5 topic alias forwarding (#2250) 2026-06-13 18:04:42 +02:00
6 changed files with 77 additions and 15 deletions
@@ -218,6 +218,8 @@ public sealed class MqttConnectedClient : IDisposable
HandleTopicAlias(publishPacket);
var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
// Topic aliases are scoped to this network connection and must not be forwarded to other clients.
applicationMessage.TopicAlias = 0;
var dispatchApplicationMessageResult =
await _sessionsManager.DispatchApplicationMessage(Id, UserName, Session.Items, applicationMessage, cancellationToken).ConfigureAwait(false);
@@ -579,4 +581,4 @@ public sealed class MqttConnectedClient : IDisposable
_logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
}
}
}
}
@@ -454,6 +454,67 @@ public sealed class MqttClient_Tests : BaseTestClass
}
}
[TestMethod]
public async Task Raise_Disconnected_On_KeepAlive_Timeout()
{
using var listener = new TcpListener(System.Net.IPAddress.Loopback, 0);
listener.Start();
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
using var testEnvironment = new TestEnvironment(TestContext);
using var client = testEnvironment.CreateClient();
var options = new MqttClientOptionsBuilder()
.WithTcpServer("127.0.0.1", port)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(5))
.WithTimeout(TimeSpan.FromSeconds(10)) // For ping timeout and others
.Build();
var disconnectedTask = new TaskCompletionSource<MqttClientDisconnectedEventArgs>();
client.DisconnectedAsync += e =>
{
disconnectedTask.TrySetResult(e);
return CompletedTask.Instance;
};
// Fake server implementation
_ = Task.Run(async () =>
{
using var clientSocket = await listener.AcceptTcpClientAsync();
await using var stream = clientSocket.GetStream();
// Read CONNECT
var buffer = new byte[1024];
_ = await stream.ReadAsync(buffer.AsMemory(), CancellationToken.None);
// Send CONNACK
var connack = new byte[] { 0x20, 0x02, 0x00, 0x00 };
await stream.WriteAsync(connack.AsMemory(), CancellationToken.None);
// Wait until client drops connection because of ping timeout
while (true)
{
var readCount = await stream.ReadAsync(buffer.AsMemory(), CancellationToken.None);
if (readCount == 0)
{
break; // connection closed
}
}
});
await client.ConnectAsync(options, CancellationToken.None);
var timeoutTask = Task.Delay(TimeSpan.FromSeconds(10));
var completedTask = await Task.WhenAny(disconnectedTask.Task, timeoutTask);
listener.Stop();
Assert.AreEqual(disconnectedTask.Task, completedTask, "DisconnectedAsync was not raised.");
var result = await disconnectedTask.Task;
Assert.IsNotNull(result);
}
[TestMethod]
public async Task Publish_QoS_1_In_ApplicationMessageReceiveHandler()
{
+2 -2
View File
@@ -114,7 +114,7 @@ public sealed class Client_Tests : BaseTestClass
Assert.IsNotNull(receivedMessage);
Assert.AreEqual(applicationMessage.Topic, receivedMessage.Topic);
Assert.AreEqual(applicationMessage.TopicAlias, receivedMessage.TopicAlias);
Assert.AreEqual(0, receivedMessage.TopicAlias);
Assert.AreEqual(applicationMessage.ContentType, receivedMessage.ContentType);
Assert.AreEqual(applicationMessage.ResponseTopic, receivedMessage.ResponseTopic);
Assert.AreEqual(applicationMessage.MessageExpiryInterval, receivedMessage.MessageExpiryInterval);
@@ -344,4 +344,4 @@ public sealed class Client_Tests : BaseTestClass
memoryStream.Position = 0;
return memoryStream;
}
}
}
@@ -34,6 +34,7 @@ public sealed class Topic_Alias_Tests : BaseTestClass
await testEnvironment.StartServer();
var receivedTopics = new List<string>();
var receivedTopicAliases = new List<ushort>();
var c1 = await testEnvironment.ConnectClient(options => options.WithProtocolVersion(MqttProtocolVersion.V500));
c1.ApplicationMessageReceivedAsync += e =>
@@ -41,6 +42,7 @@ public sealed class Topic_Alias_Tests : BaseTestClass
lock (receivedTopics)
{
receivedTopics.Add(e.ApplicationMessage.Topic);
receivedTopicAliases.Add(e.ApplicationMessage.TopicAlias);
}
return CompletedTask.Instance;
@@ -68,5 +70,6 @@ public sealed class Topic_Alias_Tests : BaseTestClass
Assert.HasCount(3, receivedTopics);
CollectionAssert.AllItemsAreNotNull(receivedTopics);
Assert.IsTrue(receivedTopics.All(t => t.Equals("this_is_the_topic", System.StringComparison.Ordinal)));
Assert.IsTrue(receivedTopicAliases.All(a => a == 0));
}
}
}
+6
View File
@@ -1049,6 +1049,12 @@ public sealed class MqttClient : Disposable, IMqttClient
switch (exception)
{
case OperationCanceledException:
// if (cancellationToken.IsCancellationRequested)
// {
// return;
// }
//break;
return;
case MqttCommunicationException:
_logger.Warning(exception, "Communication error while sending/receiving keep alive packets");
+1 -11
View File
@@ -1,11 +1 @@
* Core: Used new language features across the entire library
* Core: Performance improvements
* Core: Added validations for variable byte integers which do not match the .NET uint perfectly
* Client: Added support for pre-encoded UTF-8 binary buffers for user properties (#2228, thanks to @koepalex)
* Server: Improved performance of retained messages when no event handler is attached (#2093, thanks to @zhaowgit)
* Server: The event `InterceptingClientEnqueue` is now also called for retained messages (BREAKING CHANGE!)
* Server: The local end point is now also exposed in the channel adapter (#2179)
* Marked all projects as AOT compatible
* Restored the strong name of the nugets
* Embedded license file in all nugets (#2197, thanks to @JensNordenbro)
* Added support for dotnet10
- Server: Fixed exposinbg Topic Alias to clients (#2250, thanks to @suhashollakc)