feat(data-classes): Amazon MQ as an event source for AWS Lambda

Author: Michael Brewer

aws_lambda_powertools/utilities/data_classes/active_mq.py

@@ -0,0 +1,72 @@
+import base64
+import json
+from typing import Any, Iterator
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class ActiveMQMessage(DictWrapper):
+    @property
+    def message_id(self) -> str:
+        return self["messageID"]
+
+    @property
+    def message_type(self) -> str:
+        return self["messageType"]
+
+    @property
+    def data(self) -> str:
+        return self["data"]
+
+    @property
+    def decoded_data(self) -> str:
+        """Decodes the data as a str"""
+        return base64.b64decode(self.data.encode()).decode()
+
+    @property
+    def json_data(self) -> Any:
+        """Parses the data as json"""
+        return json.loads(self.decoded_data)
+
+    @property
+    def connection_id(self) -> str:
+        return self["connectionId"]
+
+    @property
+    def redelivered(self) -> bool:
+        return self["redelivered"]
+
+    @property
+    def timestamp(self) -> int:
+        return self["timestamp"]
+
+    @property
+    def broker_in_time(self) -> int:
+        return self["brokerInTime"]
+
+
+class ActiveMQEvent(DictWrapper):
+    """Represents an Active MQ event sent to Lambda
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/lambda/latest/dg/with-mq.html
+    - https://aws.amazon.com/blogs/compute/using-amazon-mq-as-an-event-source-for-aws-lambda/
+    """
+
+    @property
+    def event_source(self) -> str:
+        return self["eventSource"]
+
+    @property
+    def event_source_arn(self) -> str:
+        return self["eventSourceArn"]
+
+    @property
+    def messages(self) -> Iterator[ActiveMQMessage]:
+        for record in self["messages"]:
+            yield ActiveMQMessage(record)
+
+    @property
+    def message(self) -> ActiveMQMessage:
+        return next(self.messages)

tests/events/activeMQEvent.json

@@ -0,0 +1,45 @@
+{
+  "eventSource": "aws:amq",
+  "eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",
+  "messages": [
+    {
+      "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
+      "messageType": "jms/text-message",
+      "data": "QUJDOkFBQUE=",
+      "connectionId": "myJMSCoID",
+      "redelivered": false,
+      "destination": {
+        "physicalname": "testQueue"
+      },
+      "timestamp": 1598827811958,
+      "brokerInTime": 1598827811958,
+      "brokerOutTime": 1598827811959
+    },
+    {
+      "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
+      "messageType": "jms/text-message",
+      "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ==",
+      "connectionId": "myJMSCoID2",
+      "redelivered": false,
+      "destination": {
+        "physicalname": "testQueue"
+      },
+      "timestamp": 1598827811958,
+      "brokerInTime": 1598827811958,
+      "brokerOutTime": 1598827811959
+    },
+    {
+      "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
+      "messageType": "jms/bytes-message",
+      "data": "3DTOOW7crj51prgVLQaGQ82S48k=",
+      "connectionId": "myJMSCoID1",
+      "persistent": false,
+      "destination": {
+        "physicalname": "testQueue"
+      },
+      "timestamp": 1598827811958,
+      "brokerInTime": 1598827811958,
+      "brokerOutTime": 1598827811959
+    }
+  ]
+}

tests/events/rabbitMQEvent.json

@@ -0,0 +1,51 @@
+{
+  "eventSource": "aws:rmq",
+  "eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8",
+  "rmqMessagesByQueue": {
+    "pizzaQueue::/": [
+      {
+        "basicProperties": {
+          "contentType": "text/plain",
+          "contentEncoding": null,
+          "headers": {
+            "header1": {
+              "bytes": [
+                118,
+                97,
+                108,
+                117,
+                101,
+                49
+              ]
+            },
+            "header2": {
+              "bytes": [
+                118,
+                97,
+                108,
+                117,
+                101,
+                50
+              ]
+            },
+            "numberInHeader": 10
+          },
+          "deliveryMode": 1,
+          "priority": 34,
+          "correlationId": null,
+          "replyTo": null,
+          "expiration": "60000",
+          "messageId": null,
+          "timestamp": "Jan 1, 1970, 12:33:41 AM",
+          "type": null,
+          "userId": "AIDACKCEVSQ6C2EXAMPLE",
+          "appId": null,
+          "clusterId": null,
+          "bodySize": 80
+        },
+        "redelivered": false,
+        "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ=="
+      }
+    ]
+  }
+}

tests/functional/data_classes/test_amazon_mq.py

@@ -0,0 +1,25 @@
+from aws_lambda_powertools.utilities.data_classes.active_mq import ActiveMQEvent, ActiveMQMessage
+from tests.functional.utils import load_event
+
+
+def test_cloud_watch_trigger_event():
+    event = ActiveMQEvent(load_event("activeMQEvent.json"))
+
+    assert event.event_source == "aws:amq"
+    assert event.event_source_arn is not None
+    assert len(list(event.messages)) == 3
+
+    message = event.message
+    assert isinstance(message, ActiveMQMessage)
+    assert message.message_id is not None
+    assert message.message_type is not None
+    assert message.data is not None
+    assert message.decoded_data is not None
+    assert message.connection_id is not None
+    assert message.redelivered is False
+    assert message.timestamp is not None
+    assert message.broker_in_time is not None
+
+    messages = list(event.messages)
+    message = messages[1]
+    assert message.json_data["timeout"] == 0