Bug fix: Direct method subscription not removed after disconnect (#3359)

* Update MqttTransportHandler.cs

* Update MqttTransportHandler.cs

* no need to unhook disconnect handler

* Added unit test, removed ClientWasConnected check as the disconnect handler is only being triggered after disconnection

* Update MqttTransportHandlerTests.cs

* updated analyzer version

* reverting version

* updated netanalyzer

Author: tmahmood-microsoft

iothub/device/src/Microsoft.Azure.Devices.Client.csproj

@@ -60,7 +60,7 @@
     <PackageReference Include="Azure.Core" Version="1.31.0" />
     <PackageReference Include="MQTTnet" Version="4.1.4.563" />
     <PackageReference Include="Microsoft.Azure.Amqp" Version="2.6.2" />
-    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.3">
+    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.4-preview1.23354.4">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>

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,7 @@
 
 namespace Microsoft.Azure.Devices.Client.Tests.Transport.Mqtt
 {
+    [TestClass]
     public class MqttTransportHandlerTests
     {
         [TestMethod]
@@ -24,6 +26,20 @@ public async Task MqttTransportHandler_OpenAsyncCallsConnectAsync()
             mockMqttClient.Verify(p => p.ConnectAsync(It.IsAny<MqttClientOptions>(), cancellationToken));
         }
 
+        [TestMethod]
+        public async Task MqttTransportHandler_Disconnect_UnhooksMessageProcessorHandler()
+        {
+            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();

iothub/service/src/Microsoft.Azure.Devices.csproj

@@ -64,7 +64,7 @@
     <PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
     <PackageReference Include="System.Diagnostics.TraceSource" Version="4.3.0" />
     <PackageReference Include="System.Diagnostics.Contracts" Version="4.3.0" />
-    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.3">
+    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.4-preview1.23354.4">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>

provisioning/device/src/Microsoft.Azure.Devices.Provisioning.Client.csproj

@@ -65,7 +65,7 @@
     <PackageReference Include="Microsoft.Azure.Amqp" Version="2.6.2" />
     <PackageReference Include="MQTTnet" Version="4.1.4.563" />
     <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
-    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.3">
+    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.4-preview1.23354.4">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>

provisioning/service/src/Microsoft.Azure.Devices.Provisioning.Service.csproj

@@ -58,7 +58,7 @@
   <ItemGroup>
     <PackageReference Include="Azure.Core" Version="1.31.0" />
     <PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
-    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.3">
+    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.4-preview1.23354.4">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>