mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-18 10:26:11 +08:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f6357c42b6 | ||
|
|
16512d5b89 | ||
|
|
745d1d25fe | ||
|
|
905f070c37 | ||
|
|
bca2ded8eb |
@@ -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()
|
||||
{
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user