manual acks

Author: timtay-microsoft

iothub/device/src/Transport/Mqtt/MqttTransportHandler.cs

@@ -1281,7 +1281,6 @@ private static Message ProcessC2DMessage(MqttApplicationMessageReceivedEventArgs
 
         private async Task HandleReceivedMessageAsync(MqttApplicationMessageReceivedEventArgs receivedEventArgs)
         {
-            receivedEventArgs.AutoAcknowledge = false;
             string topic = receivedEventArgs.ApplicationMessage.Topic;
 
             if (topic.StartsWith(_deviceBoundMessagesTopic, StringComparison.InvariantCulture))
@@ -1300,21 +1299,25 @@ private async Task HandleReceivedMessageAsync(MqttApplicationMessageReceivedEven
             else if (topic.StartsWith(TwinDesiredPropertiesPatchTopic, StringComparison.InvariantCulture))
             {
                 HandleReceivedDesiredPropertiesUpdateRequest(receivedEventArgs);
+                await receivedEventArgs.AcknowledgeAsync(CancellationToken.None);
             }
             else if (topic.StartsWith(TwinResponseTopic, StringComparison.InvariantCulture))
             {
                 HandleTwinResponse(receivedEventArgs);
+                await receivedEventArgs.AcknowledgeAsync(CancellationToken.None);
             }
             else if (topic.StartsWith(DirectMethodsRequestTopic, StringComparison.InvariantCulture))
             {
                 HandleReceivedDirectMethodRequest(receivedEventArgs);
+                await receivedEventArgs.AcknowledgeAsync(CancellationToken.None);
             }
             else if (topic.StartsWith(_moduleEventMessageTopic, StringComparison.InvariantCulture)
                 || topic.StartsWith(_edgeModuleInputEventsTopic, StringComparison.InvariantCulture))
             {
                 // This works regardless of if the event is on a particular Edge module input or if
                 // the module is not an Edge module.
                 await HandleIncomingEventMessageAsync(receivedEventArgs).ConfigureAwait(false);
+                await receivedEventArgs.AcknowledgeAsync(CancellationToken.None);
             }
             else if (Logging.IsEnabled)
             {
@@ -1352,9 +1355,6 @@ private async Task HandleReceivedCloudToDeviceMessageAsync(Message receivedCloud
 
         private void HandleReceivedDirectMethodRequest(MqttApplicationMessageReceivedEventArgs receivedEventArgs)
         {
-            // This message is always QoS 0, so no ack will be sent.
-            receivedEventArgs.AutoAcknowledge = true;
-
             byte[] payload = receivedEventArgs.ApplicationMessage.Payload.ToArray();
 
             string[] tokens = Regex.Split(receivedEventArgs.ApplicationMessage.Topic, "/", RegexOptions.Compiled);
@@ -1372,18 +1372,13 @@ private void HandleReceivedDirectMethodRequest(MqttApplicationMessageReceivedEve
 
         private void HandleReceivedDesiredPropertiesUpdateRequest(MqttApplicationMessageReceivedEventArgs receivedEventArgs)
         {
-            // This message is always QoS 0, so no ack will be sent.
-            receivedEventArgs.AutoAcknowledge = true;
             var twinProperties = JsonConvert.DeserializeObject<TwinCollection>(Encoding.UTF8.GetString(receivedEventArgs.ApplicationMessage.Payload.ToArray()), JsonSerializerSettingsInitializer.GetJsonSerializerSettings());
 
             _onDesiredStatePatchListener.Invoke(twinProperties);
         }
 
         private void HandleTwinResponse(MqttApplicationMessageReceivedEventArgs receivedEventArgs)
         {
-            // This message is always QoS 0, so no ack will be sent.
-            receivedEventArgs.AutoAcknowledge = true;
-
             if (ParseResponseTopic(receivedEventArgs.ApplicationMessage.Topic, out string receivedRequestId, out int status, out long version))
             {
                 byte[] payloadBytes = receivedEventArgs.ApplicationMessage.Payload.ToArray() ?? Array.Empty<byte>();
@@ -1496,8 +1491,6 @@ private void RemoveOldOperations(object state)
 
         private async Task HandleIncomingEventMessageAsync(MqttApplicationMessageReceivedEventArgs receivedEventArgs)
         {
-            receivedEventArgs.AutoAcknowledge = true;
-
             using var iotHubMessage = new Message(receivedEventArgs.ApplicationMessage.Payload.ToArray());
 
             // The MqttTopic is in the format - devices/deviceId/modules/moduleId/inputs/inputName