Bug fix: Direct method subscription not removed after disconnect

iothub/device/src/Pipeline/MqttTransportHandler.cs

@@ -1069,14 +1069,13 @@ private Task HandleDisconnectionAsync(MqttClientDisconnectedEventArgs disconnect
             if (Logging.IsEnabled)
                 Logging.Info(this, $"MQTT connection was lost {disconnectedEventArgs.Exception}", nameof(HandleDisconnectionAsync));
 
-            if (disconnectedEventArgs.ClientWasConnected)
-            {
-                OnTransportDisconnected();
+            OnTransportDisconnected();
 
-                // During a disconnection, any pending twin updates won't be received, so we'll preemptively
-                // cancel these operations so the client can retry once reconnected.
-                RemoveOldOperations(TimeSpan.Zero);
-            }
+            // During a disconnection, any pending twin updates won't be received, so we'll preemptively
+            // cancel these operations so the client can retry once reconnected.
+            RemoveOldOperations(TimeSpan.Zero);
+
+            _mqttClient.ApplicationMessageReceivedAsync -= HandleReceivedMessageAsync;
 
             return Task.CompletedTask;
         }

iothub/device/tests/Transport/Mqtt/MqttTransportHandlerTests.cs

@@ -1,4 +1,5 @@
-using System.Threading;
+using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Azure.Devices.Client.Transport;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -7,6 +8,8 @@
 
 namespace Microsoft.Azure.Devices.Client.Tests.Transport.Mqtt
 {
+    [TestClass]
+    [TestCategory("Unit")]
     public class MqttTransportHandlerTests
     {
         [TestMethod]
@@ -24,6 +27,20 @@ public async Task MqttTransportHandler_OpenAsyncCallsConnectAsync()
             mockMqttClient.Verify(p => p.ConnectAsync(It.IsAny<MqttClientOptions>(), cancellationToken));
         }
 
+        [TestMethod]
+        public async Task mqtt()
+        {
+            var cancellationToken = new CancellationToken();
+            var options = new MqttClientOptions();
+
+            var mockMqttClient = new Mock<IMqttClient>();
+            using MqttTransportHandler mqttTransportHandler = CreateTransportHandler(mockMqttClient.Object);
+            await mqttTransportHandler.OpenAsync(cancellationToken);
+            mockMqttClient.VerifyAdd(p => p.ApplicationMessageReceivedAsync += It.IsAny<Func<MqttApplicationMessageReceivedEventArgs, Task>>());
+            mockMqttClient.Raise(p => p.DisconnectedAsync += null, It.IsAny<Func<MqttClientDisconnectedEventArgs, Task>>());
+            mockMqttClient.VerifyRemove(p => p.ApplicationMessageReceivedAsync -= It.IsAny<Func<MqttApplicationMessageReceivedEventArgs, Task>>());
+        }
+
         internal static MqttTransportHandler CreateTransportHandler(IMqttClient mockMqttClient)
         {
             var pipelineContext = new PipelineContext();