Fix Managed Client should dispose inner mqtt client #527

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -193,8 +193,9 @@ namespace MQTTnet.Extensions.ManagedClient
 
         public void Dispose()
         {
-            _connectionCancellationToken?.Dispose();
+            _mqttClient?.Dispose();
-            _publishingCancellationToken?.Dispose();
+            StopPublishing();
+            StopMaintainingConnection();
         }
 
         private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
@@ -355,7 +356,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
         private async Task SynchronizeSubscriptionsAsync()
         {
-            _logger.Info(nameof(ManagedMqttClient), "Synchronizing subscriptions");
+            _logger.Info("Synchronizing subscriptions");
 
             List<TopicFilter> subscriptions;
             HashSet<string> unsubscriptions;
@@ -457,4 +458,4 @@ namespace MQTTnet.Extensions.ManagedClient
             _connectionCancellationToken = null;
         }
     }
-}
+}

Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs

@@ -1,7 +1,12 @@
-using System.Threading.Tasks;
+using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Client.Options;
+using MQTTnet.Diagnostics;
 using MQTTnet.Extensions.ManagedClient;
 using MQTTnet.Server;
+using MQTTnet.Tests.Mockups;
 
 namespace MQTTnet.Tests
 {
@@ -40,5 +45,39 @@ namespace MQTTnet.Tests
                 await managedClient.StopAsync();
             }
         }
+
+        [TestMethod]
+        public async Task ManagedClients_Will_Message_Send()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                var receivedMessagesCount = 0;
+
+                var factory = new MqttFactory();
+
+                await testEnvironment.StartServerAsync();
+
+                var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("localhost", testEnvironment.ServerPort)
+                    .WithWillMessage(willMessage);
+                var dyingClient = testEnvironment.CreateClient();
+                var dyingManagedClient = new ManagedMqttClient(dyingClient, testEnvironment.ClientLogger.CreateChildLogger());
+                await dyingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder()
+                    .WithClientOptions(clientOptions)
+                    .Build());
+
+                var recievingClient = await testEnvironment.ConnectClientAsync();
+                await recievingClient.SubscribeAsync("My/last/will");
+
+                recievingClient.UseReceivedApplicationMessageHandler(context => Interlocked.Increment(ref receivedMessagesCount));
+
+                dyingManagedClient.Dispose();
+
+                await Task.Delay(1000);
+
+                Assert.AreEqual(1, receivedMessagesCount);
+            }
+        }
     }
-}
+}

Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs

@@ -29,6 +29,10 @@ namespace MQTTnet.Tests.Mockups
 
         public int ServerPort { get; set; } = 1888;
 
+        public IMqttNetLogger ServerLogger => _serverLogger;
+
+        public IMqttNetLogger ClientLogger => _clientLogger;
+
         public TestEnvironment()
         {
             _serverLogger.LogMessagePublished += (s, e) =>
@@ -56,7 +60,9 @@ namespace MQTTnet.Tests.Mockups
 
         public IMqttClient CreateClient()
         {
-            return _mqttFactory.CreateMqttClient(_clientLogger);
+            var client = _mqttFactory.CreateMqttClient(_clientLogger);
+            _clients.Add(client);
+            return client;
         }
 
         public Task<IMqttServer> StartServerAsync()
@@ -84,10 +90,9 @@ namespace MQTTnet.Tests.Mockups
 
         public async Task<IMqttClient> ConnectClientAsync(MqttClientOptionsBuilder options)
         {
-            var client = _mqttFactory.CreateMqttClient(_clientLogger);
+            var client = CreateClient();
             await client.ConnectAsync(options.WithTcpServer("localhost", ServerPort).Build());
 
-            _clients.Add(client);
             return client;
         }
 
@@ -135,4 +140,4 @@ namespace MQTTnet.Tests.Mockups
             }
         }
     }
-}
+}