fix(iot-device): Fix for retrieving message system properties over Http

abhipsaMisra · abhipsaMisra · commit d08cdaddb86b · 2020-06-24T09:18:03.000-07:00
diff --git a/e2e/test/AzureSecurityCenterForIoTSecurityMessageE2ETests.cs b/e2e/test/AzureSecurityCenterForIoTSecurityMessageE2ETests.cs
@@ -167,7 +167,7 @@ private async Task SendSingleSecurityMessageAsync(
             Client.Message testMessage = ComposeD2CSecurityTestMessage(out string eventId, out string payload, out string p1Value);
             await deviceClient.SendEventAsync(testMessage).ConfigureAwait(false);
 
-            await ValidateEventAsync(deviceId, eventId, payload, p1Value, logAnalticsTestClient).ConfigureAwait(false);
+            await ValidateEventAsync(deviceId, testMessage, eventId, payload, p1Value, logAnalticsTestClient).ConfigureAwait(false);
         }
 
         private async Task SendSingleSecurityMessageModuleAsync(
@@ -180,17 +180,18 @@ private async Task SendSingleSecurityMessageModuleAsync(
             Client.Message testMessage = ComposeD2CSecurityTestMessage(out string eventId, out string payload, out string p1Value);
             await moduleClient.SendEventAsync(testMessage).ConfigureAwait(false);
 
-            await ValidateEventAsync(deviceId, eventId, payload, p1Value, logAnalticsTestClient).ConfigureAwait(false);
+            await ValidateEventAsync(deviceId, testMessage, eventId, payload, p1Value, logAnalticsTestClient).ConfigureAwait(false);
         }
 
         private async Task ValidateEventAsync(
             string deviceId,
+            Client.Message message,
             string eventId,
             string payload,
             string p1Value,
             AzureSecurityCenterForIoTLogAnalyticsClient logAnalticsTestClient)
         {
-            bool isReceivedEventHub = EventHubTestListener.VerifyIfMessageIsReceived(deviceId, payload, p1Value, TimeSpan.FromSeconds(10));
+            bool isReceivedEventHub = EventHubTestListener.VerifyIfMessageIsReceived(deviceId, message, payload, p1Value, TimeSpan.FromSeconds(10));
             Assert.IsFalse(isReceivedEventHub, "Security message received in customer event hub.");
             bool isReceivedOms = await logAnalticsTestClient.IsRawEventExist(deviceId, eventId).ConfigureAwait(false);
             Assert.IsTrue(isReceivedOms, "Security message was not received in customer log analytics");
diff --git a/e2e/test/CombinedClientOperationsPoolAmqpTests.cs b/e2e/test/CombinedClientOperationsPoolAmqpTests.cs
@@ -122,8 +122,8 @@ private async Task DeviceCombinedClientOperationsAsync(
             // Initialize service client for service-side operations
             using ServiceClient serviceClient = ServiceClient.CreateFromConnectionString(Configuration.IoTHub.ConnectionString);
 
-            // Message payload for C2D operation
-            var messagesSent = new Dictionary<string, List<string>>();
+            // Message payload and properties for C2D operation
+            var messagesSent = new Dictionary<string, Tuple<Message, string>>();
 
             // Twin properties
             var twinPropertyMap = new Dictionary<string, List<string>>();
@@ -134,10 +134,10 @@ private async Task DeviceCombinedClientOperationsAsync(
 
                 // Send C2D Message
                 s_log.WriteLine($"{nameof(CombinedClientOperationsPoolAmqpTests)}: Send C2D for device={testDevice.Id}");
-                (Message msg, string messageId, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
+                (Message msg, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
                 using (msg)
                 {
-                    messagesSent.Add(testDevice.Id, new List<string> { payload, p1Value });
+                    messagesSent.Add(testDevice.Id, Tuple.Create(msg, payload));
                     Task sendC2dMessage = serviceClient.SendAsync(testDevice.Id, msg);
                     initOperations.Add(sendC2dMessage);
 
@@ -170,10 +170,11 @@ private async Task DeviceCombinedClientOperationsAsync(
 
                 // C2D Operation
                 s_log.WriteLine($"{nameof(CombinedClientOperationsPoolAmqpTests)}: Operation 2: Receive C2D for device={testDevice.Id}");
-                List<string> msgSent = messagesSent[testDevice.Id];
-                string payload = msgSent[0];
-                string p1Value = msgSent[1];
-                Task verifyDeviceClientReceivesMessage = MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, payload, p1Value);
+                Tuple<Message, string> msgSent = messagesSent[testDevice.Id];
+                Message msg = msgSent.Item1;
+                string payload = msgSent.Item2;
+
+                Task verifyDeviceClientReceivesMessage = MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, msg, payload);
                 clientOperations.Add(verifyDeviceClientReceivesMessage);
 
                 // Invoke direct methods
diff --git a/e2e/test/EventHubTestListener.cs b/e2e/test/EventHubTestListener.cs
@@ -42,14 +42,14 @@ private EventHubTestListener()
         }
 
         // verify required message is present in the dictionary
-        public static bool VerifyIfMessageIsReceived(string deviceId, string payload, string p1Value, TimeSpan? maxWaitTime = null)
+        public static bool VerifyIfMessageIsReceived(string deviceId, Client.Message message, string payload, string p1Value, TimeSpan? maxWaitTime = null)
         {
             if (!maxWaitTime.HasValue)
             {
                 maxWaitTime = s_maximumWaitTime;
             }
 
-            s_log.WriteLine($"Expected payload: deviceId={deviceId}; payload={payload}; property1={p1Value}");
+            s_log.WriteLine($"Expected payload: deviceId={deviceId}; messageId = {message.MessageId}, userId={message.UserId}, payload={payload}; property1={p1Value}");
 
             bool isReceived = false;
 
@@ -64,7 +64,7 @@ public static bool VerifyIfMessageIsReceived(string deviceId, string payload, st
                     continue;
                 }
 
-                isReceived = VerifyTestMessage(eventData, deviceId, p1Value);
+                isReceived = VerifyTestMessage(eventData, deviceId, message, p1Value);
             }
 
             sw.Stop();
@@ -101,7 +101,7 @@ private static string GetEventDataBody(EventData eventData)
 #endif
         }
 
-        private static bool VerifyTestMessage(EventData eventData, string deviceName, string p1Value)
+        private static bool VerifyTestMessage(EventData eventData, string deviceName, Client.Message message, string p1Value)
         {
 #if NET451
             var connectionDeviceId = eventData.SystemProperties["iothub-connection-device-id"].ToString();
diff --git a/e2e/test/FaultInjectionPoolAmqpTests.MessageReceiveFaultInjectionPoolAmqpTests.cs b/e2e/test/FaultInjectionPoolAmqpTests.MessageReceiveFaultInjectionPoolAmqpTests.cs
@@ -618,15 +618,15 @@ private async Task ReceiveMessageRecoveryPoolOverAmqp(
 
             Func<DeviceClient, TestDevice, Task> testOperation = async (deviceClient, testDevice) =>
             {
-                (Message msg, string messageId, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
+                (Message msg, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
                 _log.WriteLine($"{nameof(FaultInjectionPoolAmqpTests)}: Sending message to device {testDevice.Id}: payload='{payload}' p1Value='{p1Value}'");
                 await serviceClient.SendAsync(testDevice.Id, msg)
                 .ConfigureAwait(false);
 
                 _log.WriteLine($"{nameof(FaultInjectionPoolAmqpTests)}: Preparing to receive message for device {testDevice.Id}");
                 await deviceClient.OpenAsync()
                 .ConfigureAwait(false);
-                await MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, payload, p1Value)
+                await MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, msg, payload)
                 .ConfigureAwait(false);
             };
 
diff --git a/e2e/test/FaultInjectionPoolAmqpTests.MessageSendFaultInjectionPoolAmqpTests.cs b/e2e/test/FaultInjectionPoolAmqpTests.MessageSendFaultInjectionPoolAmqpTests.cs
@@ -866,12 +866,12 @@ private async Task SendMessageRecoveryPoolOverAmqpAsync(
                 _log.WriteLine($"{nameof(FaultInjectionPoolAmqpTests)}: Preparing to send message for device {testDevice.Id}");
                 await deviceClient.OpenAsync().ConfigureAwait(false);
 
-                (Client.Message testMessage, string messageId, string payload, string p1Value) = MessageSendE2ETests.ComposeD2cTestMessage();
+                (Client.Message testMessage, string payload, string p1Value) = MessageSendE2ETests.ComposeD2cTestMessage();
 
                 _log.WriteLine($"{nameof(FaultInjectionPoolAmqpTests)}.{testDevice.Id}: payload='{payload}' p1Value='{p1Value}'");
                 await deviceClient.SendEventAsync(testMessage).ConfigureAwait(false);
 
-                bool isReceived = EventHubTestListener.VerifyIfMessageIsReceived(testDevice.Id, payload, p1Value);
+                bool isReceived = EventHubTestListener.VerifyIfMessageIsReceived(testDevice.Id, testMessage, payload, p1Value);
                 Assert.IsTrue(isReceived, $"Message is not received for device {testDevice.Id}.");
             };
 
diff --git a/e2e/test/MessageFeedbackE2ETests.cs b/e2e/test/MessageFeedbackE2ETests.cs
@@ -58,7 +58,7 @@ private static async Task CompleteMessageMixOrder(TestDeviceType type, Client.Tr
                 var messages = new List<Client.Message>();
                 for (int i = 0; i < MESSAGE_COUNT; i++)
                 {
-                    (Message msg, string messageId, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
+                    (Message msg, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
                     await serviceClient.SendAsync(testDevice.Id, msg).ConfigureAwait(false);
                     Client.Message message = await deviceClient.ReceiveAsync(TIMESPAN_ONE_MINUTE).ConfigureAwait(false);
                     if (message == null)
diff --git a/e2e/test/MessageReceiveE2EPoolAmqpTests.cs b/e2e/test/MessageReceiveE2EPoolAmqpTests.cs
@@ -114,15 +114,15 @@ private async Task ReceiveMessagePoolOverAmqpAsync(
             int devicesCount,
             ConnectionStringAuthScope authScope = ConnectionStringAuthScope.Device)
         {
-            var messagesSent = new Dictionary<string, List<string>>();
+            var messagesSent = new Dictionary<string, Tuple<Message, string>>();
 
             // Initialize the service client
             ServiceClient serviceClient = ServiceClient.CreateFromConnectionString(Configuration.IoTHub.ConnectionString);
 
             Func<DeviceClient, TestDevice, Task> initOperation = async (deviceClient, testDevice) =>
             {
-                (Message msg, string messageId, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
-                messagesSent.Add(testDevice.Id, new List<string> { payload, p1Value });
+                (Message msg, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
+                messagesSent.Add(testDevice.Id, Tuple.Create(msg, payload));
 
                 await serviceClient.SendAsync(testDevice.Id, msg).ConfigureAwait(false);
             };
@@ -132,11 +132,8 @@ private async Task ReceiveMessagePoolOverAmqpAsync(
                 _log.WriteLine($"{nameof(MessageReceiveE2EPoolAmqpTests)}: Preparing to receive message for device {testDevice.Id}");
                 await deviceClient.OpenAsync().ConfigureAwait(false);
 
-                List<string> msgSent = messagesSent[testDevice.Id];
-                string payload = msgSent[0];
-                string p1Value = msgSent[1];
-
-                await MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, payload, p1Value).ConfigureAwait(false);
+                Tuple<Message, string> msgSent = messagesSent[testDevice.Id];
+                await MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, msgSent.Item1, msgSent.Item2).ConfigureAwait(false);
             };
 
             Func<Task> cleanupOperation = async () =>
diff --git a/e2e/test/MessageReceiveE2ETests.cs b/e2e/test/MessageReceiveE2ETests.cs
@@ -20,6 +20,7 @@ namespace Microsoft.Azure.Devices.E2ETests
     public partial class MessageReceiveE2ETests : IDisposable
     {
         private static readonly string s_devicePrefix = $"E2E_{nameof(MessageReceiveE2ETests)}_";
+
         private static readonly TestLogging s_log = TestLogging.GetInstance();
         private static readonly TimeSpan s_oneMinute = TimeSpan.FromMinutes(1);
         private static readonly TimeSpan s_oneSecond = TimeSpan.FromSeconds(1);
@@ -220,30 +221,33 @@ public async Task Message_DeviceReceiveMessageOperationTimeout_MqttWs()
             await ReceiveMessageInOperationTimeoutAsync(TestDeviceType.Sasl, Client.TransportType.Mqtt_WebSocket_Only).ConfigureAwait(false);
         }
 
-        public static (Message message, string messageId, string payload, string p1Value) ComposeC2dTestMessage()
+        public static (Message message, string payload, string p1Value) ComposeC2dTestMessage()
         {
             var payload = Guid.NewGuid().ToString();
             var messageId = Guid.NewGuid().ToString();
             var p1Value = Guid.NewGuid().ToString();
+            var userId = Guid.NewGuid().ToString();
 
-            s_log.WriteLine($"{nameof(ComposeC2dTestMessage)}: messageId='{messageId}' payload='{payload}' p1Value='{p1Value}'");
+            s_log.WriteLine($"{nameof(ComposeC2dTestMessage)}: messageId='{messageId}' userId='{userId}' payload='{payload}' p1Value='{p1Value}'");
             var message = new Message(Encoding.UTF8.GetBytes(payload))
             {
                 MessageId = messageId,
+                UserId = userId,
                 Properties = { ["property1"] = p1Value }
             };
 
-            return (message, messageId, payload, p1Value);
+            return (message, payload, p1Value);
         }
 
-        public static async Task VerifyReceivedC2DMessageAsync(Client.TransportType transport, DeviceClient dc, string deviceId, string payload, string p1Value)
+        public static async Task VerifyReceivedC2DMessageAsync(Client.TransportType transport, DeviceClient dc, string deviceId, Message message, string payload)
         {
+            string receivedMessageDestination = $"/devices/{deviceId}/messages/deviceBound";
+
             var sw = new Stopwatch();
             bool received = false;
 
             sw.Start();
 
-
             while (!received && sw.ElapsedMilliseconds < FaultInjection.RecoveryTimeMilliseconds)
             {
                 Client.Message receivedMessage = null;
@@ -277,14 +281,19 @@ public static async Task VerifyReceivedC2DMessageAsync(Client.TransportType tran
                         // ignore exception from CompleteAsync
                     }
 
+                    Assert.AreEqual(receivedMessage.MessageId, message.MessageId, "Recieved message Id is not what was sent by service");
+                    Assert.AreEqual(receivedMessage.UserId, message.UserId, "Recieved user Id is not what was sent by service");
+                    Assert.AreEqual(receivedMessage.To, receivedMessageDestination, "Recieved message destination is not what was sent by service");
+
                     string messageData = Encoding.ASCII.GetString(receivedMessage.GetBytes());
                     s_log.WriteLine($"{nameof(VerifyReceivedC2DMessageAsync)}: Received message: for {deviceId}: {messageData}");
                     if (Equals(payload, messageData))
                     {
                         Assert.AreEqual(1, receivedMessage.Properties.Count, $"The count of received properties did not match for device {deviceId}");
                         System.Collections.Generic.KeyValuePair<string, string> prop = receivedMessage.Properties.Single();
-                        Assert.AreEqual("property1", prop.Key, $"The key \"property1\" did not match for device {deviceId}");
-                        Assert.AreEqual(p1Value, prop.Value, $"The value of \"property1\" did not match for device {deviceId}");
+                        string propertyKey = "property1";
+                        Assert.AreEqual(propertyKey, prop.Key, $"The key \"property1\" did not match for device {deviceId}");
+                        Assert.AreEqual(message.Properties[propertyKey], prop.Value, $"The value of \"property1\" did not match for device {deviceId}");
                         received = true;
                     }
                 }
@@ -422,11 +431,11 @@ private async Task ReceiveSingleMessageAsync(TestDeviceType type, Client.Transpo
 
             await serviceClient.OpenAsync().ConfigureAwait(false);
 
-            (Message msg, string messageId, string payload, string p1Value) = ComposeC2dTestMessage();
+            (Message msg, string payload, string p1Value) = ComposeC2dTestMessage();
             using (msg)
             {
                 await serviceClient.SendAsync(testDevice.Id, msg).ConfigureAwait(false);
-                await VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, payload, p1Value).ConfigureAwait(false);
+                await VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, msg, payload).ConfigureAwait(false);
             }
 
             await deviceClient.CloseAsync().ConfigureAwait(false);
@@ -450,7 +459,7 @@ private async Task ReceiveSingleMessageWithCancellationTokenAsync(TestDeviceType
 
             await serviceClient.OpenAsync().ConfigureAwait(false);
 
-            (Message msg, string messageId, string payload, string p1Value) = ComposeC2dTestMessage();
+            (Message msg, string payload, string p1Value) = ComposeC2dTestMessage();
             using (msg)
             {
                 await serviceClient.SendAsync(testDevice.Id, msg).ConfigureAwait(false);
diff --git a/e2e/test/MessageReceiveFaultInjectionTests.cs b/e2e/test/MessageReceiveFaultInjectionTests.cs
@@ -204,9 +204,9 @@ private async Task ReceiveMessageRecovery(
 
                 Func<DeviceClient, TestDevice, Task> testOperation = async (deviceClient, testDevice) =>
                 {
-                    (Message message, string messageId, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
+                    (Message message, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();
                     await serviceClient.SendAsync(testDevice.Id, message).ConfigureAwait(false);
-                    await MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, payload, p1Value).ConfigureAwait(false);
+                    await MessageReceiveE2ETests.VerifyReceivedC2DMessageAsync(transport, deviceClient, testDevice.Id, message, payload).ConfigureAwait(false);
                 };
 
                 Func<Task> cleanupOperation = () =>
diff --git a/e2e/test/MessageSendE2EPoolAmqpTests.cs b/e2e/test/MessageSendE2EPoolAmqpTests.cs
@@ -130,11 +130,11 @@ private async Task SendMessagePoolOverAmqp(
                 s_log.WriteLine($"{nameof(MessageSendE2EPoolAmqpTests)}: Preparing to send message for device {testDevice.Id}");
                 await deviceClient.OpenAsync().ConfigureAwait(false);
 
-                (Client.Message testMessage, string messageId, string payload, string p1Value) = MessageSendE2ETests.ComposeD2cTestMessage();
-                s_log.WriteLine($"{nameof(MessageSendE2EPoolAmqpTests)}.{testDevice.Id}: messageId='{messageId}' payload='{payload}' p1Value='{p1Value}'");
+                (Client.Message testMessage, string payload, string p1Value) = MessageSendE2ETests.ComposeD2cTestMessage();
+                s_log.WriteLine($"{nameof(MessageSendE2EPoolAmqpTests)}.{testDevice.Id}: messageId='{testMessage.MessageId}' payload='{payload}' p1Value='{p1Value}'");
                 await deviceClient.SendEventAsync(testMessage).ConfigureAwait(false);
 
-                bool isReceived = EventHubTestListener.VerifyIfMessageIsReceived(testDevice.Id, payload, p1Value);
+                bool isReceived = EventHubTestListener.VerifyIfMessageIsReceived(testDevice.Id, testMessage, payload, p1Value);
                 Assert.IsTrue(isReceived, "Message is not received.");
             };
 
diff --git a/e2e/test/MessageSendE2ETests.cs b/e2e/test/MessageSendE2ETests.cs
diff --git a/e2e/test/MessageSendFaultInjectionTests.cs b/e2e/test/MessageSendFaultInjectionTests.cs
diff --git a/iothub/device/src/Transport/HttpTransportHandler.cs b/iothub/device/src/Transport/HttpTransportHandler.cs

Original file line number	Diff line number	Diff line change
`@@ -42,14 +42,14 @@ private EventHubTestListener()`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`// verify required message is present in the dictionary`
`45`		`- public static bool VerifyIfMessageIsReceived(string deviceId, string payload, string p1Value, TimeSpan? maxWaitTime = null)`
	`45`	`+ public static bool VerifyIfMessageIsReceived(string deviceId, Client.Message message, string payload, string p1Value, TimeSpan? maxWaitTime = null)`
`46`	`46`	`{`
`47`	`47`	`if (!maxWaitTime.HasValue)`
`48`	`48`	`{`
`49`	`49`	`maxWaitTime = s_maximumWaitTime;`
`50`	`50`	`}`
`51`	`51`
`52`		`- s_log.WriteLine($"Expected payload: deviceId={deviceId}; payload={payload}; property1={p1Value}");`
	`52`	`+ s_log.WriteLine($"Expected payload: deviceId={deviceId}; messageId = {message.MessageId}, userId={message.UserId}, payload={payload}; property1={p1Value}");`
`53`	`53`
`54`	`54`	`bool isReceived = false;`
`55`	`55`
`@@ -64,7 +64,7 @@ public static bool VerifyIfMessageIsReceived(string deviceId, string payload, st`
`64`	`64`	`continue;`
`65`	`65`	`}`
`66`	`66`
`67`		`- isReceived = VerifyTestMessage(eventData, deviceId, p1Value);`
	`67`	`+ isReceived = VerifyTestMessage(eventData, deviceId, message, p1Value);`
`68`	`68`	`}`
`69`	`69`
`70`	`70`	`sw.Stop();`
`@@ -101,7 +101,7 @@ private static string GetEventDataBody(EventData eventData)`
`101`	`101`	`#endif`
`102`	`102`	`}`
`103`	`103`
`104`		`- private static bool VerifyTestMessage(EventData eventData, string deviceName, string p1Value)`
	`104`	`+ private static bool VerifyTestMessage(EventData eventData, string deviceName, Client.Message message, string p1Value)`
`105`	`105`	`{`
`106`	`106`	`#if NET451`
`107`	`107`	`var connectionDeviceId = eventData.SystemProperties["iothub-connection-device-id"].ToString();`
Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ private static async Task CompleteMessageMixOrder(TestDeviceType type, Client.Tr`
`58`	`58`	`var messages = new List<Client.Message>();`
`59`	`59`	`for (int i = 0; i < MESSAGE_COUNT; i++)`
`60`	`60`	`{`
`61`		`- (Message msg, string messageId, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();`
	`61`	`+ (Message msg, string payload, string p1Value) = MessageReceiveE2ETests.ComposeC2dTestMessage();`
`62`	`62`	`await serviceClient.SendAsync(testDevice.Id, msg).ConfigureAwait(false);`
`63`	`63`	`Client.Message message = await deviceClient.ReceiveAsync(TIMESPAN_ONE_MINUTE).ConfigureAwait(false);`
`64`	`64`	`if (message == null)`