Add connection deviceId and moduleId to incoming messages on Amqp

iothub/device/src/Message.cs

@@ -436,6 +436,11 @@ public string ConnectionDeviceId
                 return this.GetSystemProperty<string>(MessageSystemPropertyNames.ConnectionDeviceId);
 #endif
             }
+
+            internal set
+            {
+                this.SystemProperties[MessageSystemPropertyNames.ConnectionDeviceId] = value;
+            }
         }
 
         /// <summary>
@@ -451,6 +456,11 @@ public string ConnectionModuleId
                 return this.GetSystemProperty<string>(MessageSystemPropertyNames.ConnectionModuleId);
 #endif
             }
+
+            internal set
+            {
+                this.SystemProperties[MessageSystemPropertyNames.ConnectionModuleId] = value;
+            }
         }
 
         /// <summary>

iothub/device/src/MessageConverter.cs

@@ -58,41 +58,27 @@ public static void UpdateMessageHeaderAndProperties(AmqpMessage amqpMessage, Mes
                     data.ContentEncoding = amqpMessage.Properties.ContentEncoding.Value;
                 }
 
-                data.UserId = amqpMessage.Properties.UserId.Array != null ? Encoding.UTF8.GetString(amqpMessage.Properties.UserId.Array, 0 /*index*/, amqpMessage.Properties.UserId.Array.Length) : null;
-
-                if (!string.IsNullOrWhiteSpace(amqpMessage.Properties.ContentType.Value))
-                {
-                    data.ContentType = amqpMessage.Properties.ContentType.Value;
-                }
-
-                if (!string.IsNullOrWhiteSpace(amqpMessage.Properties.ContentEncoding.Value))
-                {
-                    data.ContentEncoding = amqpMessage.Properties.ContentEncoding.Value;
-                }
+                data.UserId = amqpMessage.Properties.UserId.Array != null ? Encoding.UTF8.GetString(amqpMessage.Properties.UserId.Array, 0 /*index*/, amqpMessage.Properties.UserId.Array.Length) : null;                
             }
 
             if ((sections & SectionFlag.MessageAnnotations) != 0)
-            {
-                string lockToken;
-                if (amqpMessage.MessageAnnotations.Map.TryGetValue(LockTokenName, out lockToken))
+            {                
+                if (amqpMessage.MessageAnnotations.Map.TryGetValue(LockTokenName, out string lockToken))
                 {
                     data.LockToken = lockToken;
                 }
 
-                ulong sequenceNumber;
-                if (amqpMessage.MessageAnnotations.Map.TryGetValue(SequenceNumberName, out sequenceNumber))
+                if (amqpMessage.MessageAnnotations.Map.TryGetValue(SequenceNumberName, out ulong sequenceNumber))
                 {
                     data.SequenceNumber = sequenceNumber;
                 }
 
-                DateTime enqueuedTime;
-                if (amqpMessage.MessageAnnotations.Map.TryGetValue(MessageSystemPropertyNames.EnqueuedTime, out enqueuedTime))
+                if (amqpMessage.MessageAnnotations.Map.TryGetValue(MessageSystemPropertyNames.EnqueuedTime, out DateTime enqueuedTime))
                 {
                     data.EnqueuedTimeUtc = enqueuedTime;
                 }
-
-                byte deliveryCount;
-                if (amqpMessage.MessageAnnotations.Map.TryGetValue(MessageSystemPropertyNames.DeliveryCount, out deliveryCount))
+
+                if (amqpMessage.MessageAnnotations.Map.TryGetValue(MessageSystemPropertyNames.DeliveryCount, out byte deliveryCount))
                 {
                     data.DeliveryCount = deliveryCount;
                 }
@@ -101,6 +87,16 @@ public static void UpdateMessageHeaderAndProperties(AmqpMessage amqpMessage, Mes
                 {
                     data.InputName = inputName;
                 }
+
+                if (amqpMessage.MessageAnnotations.Map.TryGetValue(MessageSystemPropertyNames.ConnectionDeviceId, out string connectionDeviceId))
+                {
+                    data.ConnectionDeviceId = connectionDeviceId;
+                }
+
+                if (amqpMessage.MessageAnnotations.Map.TryGetValue(MessageSystemPropertyNames.ConnectionModuleId, out string connectionModuleId))
+                {
+                    data.ConnectionModuleId = connectionModuleId;
+                }
             }
 
             if ((sections & SectionFlag.ApplicationProperties) != 0)

iothub/device/src/MessageSystemPropertyNames.cs

@@ -39,9 +39,9 @@ static class MessageSystemPropertyNames
 
         public const string ContentType = "iothub-content-type";
 
-        public const string ConnectionDeviceId = "connectionDeviceId";
+        public const string ConnectionDeviceId = "iothub-connection-device-id";
 
-        public const string ConnectionModuleId = "connectionModuleId";
+        public const string ConnectionModuleId = "iothub-connection-module-id";
 
         public const string DiagId = "iothub-diag-id";
 

iothub/device/tests/MessageConverterTests.cs

@@ -0,0 +1,70 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using Microsoft.Azure.Amqp;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Microsoft.Azure.Devices.Client.Test
+{
+    [TestClass]
+    public class MessageConverterTests
+    {
+        [TestCategory("CIT")]
+        [TestMethod]
+        public void UpdateMessageHeaderAndPropertiesTest()
+        {
+            byte[] bytes = { 1, 2, 3, 4 };
+            string messageId = Guid.NewGuid().ToString();
+            string correlationId = Guid.NewGuid().ToString();
+            string contentType = "application/json";
+            string contentEncoding = "UTF-8";
+            string to = "d1";
+            var enqueuedTime = new DateTime(2018, 4, 5, 04, 05, 06, DateTimeKind.Utc);
+            byte deliveryCount = 10;
+            string messageSchema = "testSchema";
+            string connectionDeviceId = "connD1";
+            string connectionModuleId = "connM1";
+
+            using (AmqpMessage amqpMessage =
+                AmqpMessage.Create(new Amqp.Framing.Data { Value = new ArraySegment<byte>(bytes) }))
+            {
+                amqpMessage.Properties.MessageId = messageId;
+                amqpMessage.Properties.CorrelationId = correlationId;
+                amqpMessage.Properties.ContentType = contentType;
+                amqpMessage.Properties.ContentEncoding = contentEncoding;
+                amqpMessage.Properties.To = to;
+
+                amqpMessage.MessageAnnotations.Map[MessageSystemPropertyNames.EnqueuedTime] = enqueuedTime;
+                amqpMessage.MessageAnnotations.Map[MessageSystemPropertyNames.DeliveryCount] = deliveryCount;
+                amqpMessage.MessageAnnotations.Map[MessageSystemPropertyNames.ConnectionDeviceId] = connectionDeviceId;
+                amqpMessage.MessageAnnotations.Map[MessageSystemPropertyNames.ConnectionModuleId] = connectionModuleId;
+
+                amqpMessage.ApplicationProperties.Map[MessageSystemPropertyNames.MessageSchema] = messageSchema;
+
+                amqpMessage.ApplicationProperties.Map["Prop1"] = "Value1";
+                amqpMessage.ApplicationProperties.Map["Prop2"] = "Value2";
+
+                var message = new Message(bytes);
+
+                MessageConverter.UpdateMessageHeaderAndProperties(amqpMessage, message);
+
+                Assert.AreEqual(messageId, message.MessageId);
+                Assert.AreEqual(correlationId, message.CorrelationId);
+                Assert.AreEqual(contentType, message.ContentType);
+                Assert.AreEqual(contentEncoding, message.ContentEncoding);
+                Assert.AreEqual(to, message.To);
+
+                Assert.AreEqual(enqueuedTime, message.EnqueuedTimeUtc);
+                Assert.AreEqual(deliveryCount, message.DeliveryCount);
+                Assert.AreEqual(connectionDeviceId, message.ConnectionDeviceId);
+                Assert.AreEqual(connectionModuleId, message.ConnectionModuleId);
+
+                Assert.AreEqual(messageSchema, message.MessageSchema);
+
+                Assert.AreEqual("Value1", message.Properties["Prop1"]);
+                Assert.AreEqual("Value2", message.Properties["Prop2"]);
+            }
+        }
+    }
+}

versions.csv

@@ -1,5 +1,5 @@
 AssemblyPath, Version
-iothub\device\src\Microsoft.Azure.Devices.Client.csproj, 1.17.0
+iothub\device\src\Microsoft.Azure.Devices.Client.csproj, 1.17.1
 iothub\service\src\Microsoft.Azure.Devices.csproj, 1.16.0
 shared\src\Microsoft.Azure.Devices.Shared.csproj, 1.15.0
 provisioning\device\src\Microsoft.Azure.Devices.Provisioning.Client.csproj, 1.1.0