Fix Unit Tests.

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();

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)
             {

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)
                 {

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<MqttApplicationMessage>();
                 var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetLogger());
                 managedClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c =>
@@ -350,9 +353,9 @@ namespace MQTTnet.Tests
         }
 
         async Task<ManagedMqttClient> 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<List<MqttApplicationMessage>> SetupReceivingOfMessages(ManagedMqttClient managedClient, int expectedNumberOfMessages)
         {
             var receivedMessages = new List<MqttApplicationMessage>();
-            var allReceived = new TaskCompletionSource<List<MqttApplicationMessage>>();
+            var result = new TaskCompletionSource<List<MqttApplicationMessage>>(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;
         }
     }
 

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();

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);
             }
         }