diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 65229c97..714c24a8 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -228,6 +228,7 @@ namespace MQTTnet.Extensions.ManagedClient _unsubscriptions.Remove(topicFilter.Topic); } } + _subscriptionsQueuedSignal.Release(); return Task.FromResult(0); @@ -305,7 +306,9 @@ namespace MQTTnet.Extensions.ManagedClient _logger.Info("Stopped"); } + _reconnectSubscriptions.Clear(); + lock (_subscriptions) { _subscriptions.Clear(); diff --git a/Tests/MQTTnet.Core.Tests/Extension_Tests.cs b/Tests/MQTTnet.Core.Tests/Extension_Tests.cs index 0eaa5e0e..68951000 100644 --- a/Tests/MQTTnet.Core.Tests/Extension_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Extension_Tests.cs @@ -79,6 +79,8 @@ namespace MQTTnet.Tests [TestMethod] public async Task TimeoutAfterMemoryUsage() { + var initialMemory = GC.GetTotalMemory(true); + var tasks = Enumerable.Range(0, 100000) .Select(i => { @@ -87,10 +89,10 @@ namespace MQTTnet.Tests }); await Task.WhenAll(tasks); - AssertIsLess(3_000_000, GC.GetTotalMemory(true)); + AssertIsLess(initialMemory + 2_000_000, GC.GetTotalMemory(true)); } - private static void AssertIsLess(long bound, long actual) + static void AssertIsLess(long bound, long actual) { if (bound < actual) { diff --git a/Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs index 7aa9cd37..4bc9c43b 100644 --- a/Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs @@ -109,6 +109,7 @@ namespace MQTTnet.Tests try { await client.SendAsync(MqttPingReqPacket.Instance, CancellationToken.None).ConfigureAwait(false); + await client.SendAsync(MqttPingReqPacket.Instance, CancellationToken.None).ConfigureAwait(false); } catch (MqttCommunicationException exception) { diff --git a/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs index a9a8bfdd..ae671de7 100644 --- a/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs @@ -59,7 +59,7 @@ namespace MQTTnet.Tests using (var testEnvironment = new TestEnvironment(TestContext)) { var receivedMessagesCount = 0; - + await testEnvironment.StartServerAsync(); var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build(); @@ -270,14 +270,17 @@ namespace MQTTnet.Tests var connectionCheckInterval = TimeSpan.FromSeconds(10); var managedClient = await CreateManagedClientAsync(testEnvironment, null, connectionCheckInterval); var sendingClient = await testEnvironment.ConnectClientAsync(); + await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", Payload = new byte[] { 1 }, Retain = true }); - await managedClient.SubscribeAsync("topic"); - var subscribeTime = DateTime.UtcNow; - var messages = await SetupReceivingOfMessages(managedClient, 1); + var messagesTask = SetupReceivingOfMessages(managedClient, 1); + await managedClient.SubscribeAsync("topic"); + + var messages = await messagesTask; + var elapsed = DateTime.UtcNow - subscribeTime; Assert.IsTrue(elapsed < TimeSpan.FromSeconds(1), $"Subscriptions must be activated immediately, this one took {elapsed}"); Assert.AreEqual(messages.Single().Topic, "topic"); @@ -308,7 +311,7 @@ namespace MQTTnet.Tests var clientOptions = new MqttClientOptionsBuilder() .WithTcpServer("localhost", testEnvironment.ServerPort); - + var receivedManagedMessages = new List(); var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetLogger()); managedClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c => @@ -350,9 +353,9 @@ namespace MQTTnet.Tests } async Task CreateManagedClientAsync( - TestEnvironment testEnvironment, - IMqttClient underlyingClient = null, - TimeSpan? connectionCheckInterval = null) + TestEnvironment testEnvironment, + IMqttClient underlyingClient = null, + TimeSpan? connectionCheckInterval = null) { await testEnvironment.StartServerAsync(); @@ -368,8 +371,7 @@ namespace MQTTnet.Tests // at connection check intervals managedOptions.ConnectionCheckInterval = connectionCheckInterval ?? TimeSpan.FromSeconds(0.1); - var managedClient = - new ManagedMqttClient(underlyingClient ?? testEnvironment.CreateClient(), new MqttNetLogger()); + var managedClient = new ManagedMqttClient(underlyingClient ?? testEnvironment.CreateClient(), new MqttNetLogger()); var connected = GetConnectedTask(managedClient); @@ -401,17 +403,19 @@ namespace MQTTnet.Tests Task> SetupReceivingOfMessages(ManagedMqttClient managedClient, int expectedNumberOfMessages) { var receivedMessages = new List(); - var allReceived = new TaskCompletionSource>(); + var result = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); + managedClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(r => { receivedMessages.Add(r.ApplicationMessage); + if (receivedMessages.Count == expectedNumberOfMessages) { - allReceived.SetResult(receivedMessages); + result.TrySetResult(receivedMessages); } }); - return allReceived.Task; + return result.Task; } } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 433c2a8e..e2d15fd1 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -188,8 +188,18 @@ namespace MQTTnet.Tests.Mockups { foreach (var mqttClient in _clients) { - mqttClient.DisconnectAsync().GetAwaiter().GetResult(); - mqttClient?.Dispose(); + try + { + mqttClient.DisconnectAsync().GetAwaiter().GetResult(); + } + catch (ObjectDisposedException) + { + // This can happen when the test already disconnected the client. + } + finally + { + mqttClient?.Dispose(); + } } Server?.StopAsync().GetAwaiter().GetResult(); diff --git a/Tests/MQTTnet.Core.Tests/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_Tests.cs index 662fb634..3df77510 100644 --- a/Tests/MQTTnet.Core.Tests/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server_Tests.cs @@ -415,33 +415,27 @@ namespace MQTTnet.Tests public async Task Publish_Multiple_Clients() { var receivedMessagesCount = 0; - var locked = new object(); - + using (var testEnvironment = new TestEnvironment(TestContext)) { await testEnvironment.StartServerAsync(); var c1 = await testEnvironment.ConnectClientAsync(); var c2 = await testEnvironment.ConnectClientAsync(); - - c1.UseApplicationMessageReceivedHandler(c => - { - lock (locked) - { - receivedMessagesCount++; - } - }); + var c3 = await testEnvironment.ConnectClientAsync(); c2.UseApplicationMessageReceivedHandler(c => { - lock (locked) - { - receivedMessagesCount++; - } + Interlocked.Increment(ref receivedMessagesCount); + }); + + c3.UseApplicationMessageReceivedHandler(c => + { + Interlocked.Increment(ref receivedMessagesCount); }); - await c1.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }).ConfigureAwait(false); await c2.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }).ConfigureAwait(false); + await c3.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }).ConfigureAwait(false); var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); @@ -450,8 +444,8 @@ namespace MQTTnet.Tests await c1.PublishAsync(message).ConfigureAwait(false); } - await Task.Delay(5000); - + SpinWait.SpinUntil(() => receivedMessagesCount == 1000, TimeSpan.FromSeconds(10)); + Assert.AreEqual(1000, receivedMessagesCount); } }