[Event Hubs] Put uamqp.Message.properties into EventData.system_properties

sdk/eventhub/azure-eventhub/azure/eventhub/_common.py

@@ -29,6 +29,19 @@
     PROP_PARTITION_KEY,
     PROP_PARTITION_KEY_AMQP_SYMBOL,
     PROP_TIMESTAMP,
+    PROP_ABSOLUTE_EXPIRY_TIME,
+    PROP_CONTENT_ENCODING,
+    PROP_CONTENT_TYPE,
+    PROP_CORRELATION_ID,
+    PROP_GROUP_ID,
+    PROP_GROUP_SEQUENCE,
+    PROP_MESSAGE_ID,
+    PROP_REPLY_TO,
+    PROP_REPLY_TO_GROUP_ID,
+    PROP_SUBJECT,
+    PROP_TO,
+    PROP_USER_ID,
+    PROP_CREATION_TIME,
 )
 
 if TYPE_CHECKING:
@@ -60,6 +73,7 @@ class EventData(object):
     def __init__(self, body=None):
         # type: (Union[str, bytes, List[AnyStr]]) -> None
         self._last_enqueued_event_properties = {}  # type: Dict[str, Any]
+        self._sys_properties = None  # type: Optional[Dict[bytes, Any]]
         if body and isinstance(body, list):
             self.message = Message(body[0])
             for more in body[1:]:
@@ -207,12 +221,30 @@ def properties(self, value):
 
     @property
     def system_properties(self):
-        # type: () -> Dict[Union[str, bytes], Any]
+        # type: () -> Dict[bytes, Any]
         """Metadata set by the Event Hubs Service associated with the event
 
         :rtype: dict
         """
-        return self.message.annotations
+
+        if not self._sys_properties:
+            self._sys_properties = {
+                PROP_MESSAGE_ID: self.message.properties.message_id,
+                PROP_USER_ID: self.message.properties.user_id,
+                PROP_TO: self.message.properties.to,
+                PROP_SUBJECT: self.message.properties.subject,
+                PROP_REPLY_TO: self.message.properties.reply_to,
+                PROP_CORRELATION_ID: self.message.properties.correlation_id,
+                PROP_CONTENT_TYPE: self.message.properties.content_type,
+                PROP_CONTENT_ENCODING: self.message.properties.content_encoding,
+                PROP_ABSOLUTE_EXPIRY_TIME: self.message.properties.absolute_expiry_time,
+                PROP_CREATION_TIME: self.message.properties.creation_time,
+                PROP_GROUP_ID: self.message.properties.group_id,
+                PROP_GROUP_SEQUENCE: self.message.properties.group_sequence,
+                PROP_REPLY_TO_GROUP_ID: self.message.properties.reply_to_group_id,
+            }
+            self._sys_properties.update(self.message.annotations)
+        return self._sys_properties
 
     @property
     def body(self):

sdk/eventhub/azure-eventhub/azure/eventhub/_constants.py

@@ -17,6 +17,20 @@
 PROP_LAST_ENQUEUED_TIME_UTC = b"last_enqueued_time_utc"
 PROP_RUNTIME_INFO_RETRIEVAL_TIME_UTC = b"runtime_info_retrieval_time_utc"
 
+PROP_MESSAGE_ID = b"message-id"
+PROP_USER_ID = b"user-id"
+PROP_TO = b"to"
+PROP_SUBJECT = b"subject"
+PROP_REPLY_TO = b"reply-to"
+PROP_CORRELATION_ID = b"correlation-id"
+PROP_CONTENT_TYPE = b"content-type"
+PROP_CONTENT_ENCODING = b"content-encoding"
+PROP_ABSOLUTE_EXPIRY_TIME = b"absolute-expiry-time"
+PROP_CREATION_TIME = b"creation-time"
+PROP_GROUP_ID = b"group-id"
+PROP_GROUP_SEQUENCE = b"group-sequence"
+PROP_REPLY_TO_GROUP_ID = b"reply-to-group_id"
+
 EPOCH_SYMBOL = b"com.microsoft:epoch"
 TIMEOUT_SYMBOL = b"com.microsoft:timeout"
 RECEIVER_RUNTIME_METRIC_SYMBOL = b"com.microsoft:enable-receiver-runtime-metric"

sdk/eventhub/azure-eventhub/tests/unittest/test_event_data.py

@@ -1,6 +1,7 @@
 import platform
 import pytest
-
+import uamqp
+from azure.eventhub import _common
 
 pytestmark = pytest.mark.skipif(platform.python_implementation() == "PyPy", reason="This is ignored for PyPy")
 
@@ -52,6 +53,41 @@ def test_app_properties():
     assert event_data.properties["a"] == "b"
 
 
+def test_sys_properties():
+    properties = uamqp.message.MessageProperties()
+    properties.message_id = "message_id"
+    properties.user_id = "user_id"
+    properties.to = "to"
+    properties.subject = "subject"
+    properties.reply_to = "reply_to"
+    properties.correlation_id = "correlation_id"
+    properties.content_type = "content_type"
+    properties.content_encoding = "content_encoding"
+    properties.absolute_expiry_time = 1
+    properties.creation_time = 1
+    properties.group_id = "group_id"
+    properties.group_sequence = 1
+    properties.reply_to_group_id = "reply_to_group_id"
+    message = uamqp.Message(properties=properties)
+    message.annotations = {_common.PROP_OFFSET: "@latest"}
+    ed = EventData._from_message(message)  # type: EventData
+
+    assert ed.system_properties[_common.PROP_OFFSET] == "@latest"
+    assert ed.system_properties[_common.PROP_CORRELATION_ID] == properties.correlation_id
+    assert ed.system_properties[_common.PROP_MESSAGE_ID] == properties.message_id
+    assert ed.system_properties[_common.PROP_CONTENT_ENCODING] == properties.content_encoding
+    assert ed.system_properties[_common.PROP_CONTENT_TYPE] == properties.content_type
+    assert ed.system_properties[_common.PROP_USER_ID] == properties.user_id
+    assert ed.system_properties[_common.PROP_TO] == properties.to
+    assert ed.system_properties[_common.PROP_SUBJECT] == properties.subject
+    assert ed.system_properties[_common.PROP_REPLY_TO] == properties.reply_to
+    assert ed.system_properties[_common.PROP_ABSOLUTE_EXPIRY_TIME] == properties.absolute_expiry_time
+    assert ed.system_properties[_common.PROP_CREATION_TIME] == properties.creation_time
+    assert ed.system_properties[_common.PROP_GROUP_ID] == properties.group_id
+    assert ed.system_properties[_common.PROP_GROUP_SEQUENCE] == properties.group_sequence
+    assert ed.system_properties[_common.PROP_REPLY_TO_GROUP_ID] == properties.reply_to_group_id
+
+
 def test_event_data_batch():
     batch = EventDataBatch(max_size_in_bytes=100, partition_key="par")
     batch.add(EventData("A"))