feat(triggers): adding support for SNS -> SQS triggers (#23)

src/events/aws_sdk.js

@@ -12,6 +12,7 @@ const serverlessEvent = require('../proto/event_pb.js');
 const eventInterface = require('../event.js');
 const errorCode = require('../proto/error_code_pb.js');
 const { STEP_ID_NAME } = require('../consts.js');
+const resourceUtils = require('../resource_utils/sqs_utils.js');
 
 const s3EventCreator = {
     /**
@@ -192,6 +193,10 @@ const SQSEventCreator = {
                     'Message ID': `${response.data.Messages[0].MessageId}`,
                     'MD5 Of Message Body': `${response.data.Messages[0].MD5OfBody}`,
                 });
+                const snsData = resourceUtils.getSNSTrigger(response.data.Messages);
+                if (snsData != null) {
+                    eventInterface.addToMetadata(event, { 'SNS Trigger': snsData });
+                }
             }
             eventInterface.addToMetadata(event, { 'Number Of Messages': messagesNumber });
             break;

src/resource_utils/sqs_utils.js

@@ -0,0 +1,44 @@
+/**
+ * @fileoverview SQS utility functions
+ */
+
+/**
+ * If exists, gets the SNS message that triggered the SQS,
+ * and generates event data out of it.
+ * @param {object} messages The SQS messages object
+ * @returns {object} SNS event data json
+ */
+function getSNSTrigger(messages) {
+    let foundSnsEvent = null;
+    messages.some((message) => {
+        try {
+            let body = null;
+            if ('Body' in message) {
+                body = JSON.parse(message.Body);
+            } else if ('body' in message) {
+                body = JSON.parse(message.body);
+            } else {
+                return true;
+            }
+
+            if ('Type' in body &&
+                'MessageId' in body &&
+                'TopicArn' in body &&
+                'Message' in body &&
+                'Timestamp' in body &&
+                'SignatureVersion' in body &&
+                'Signature' in body) {
+                foundSnsEvent = body;
+                return true;
+            }
+        } catch (ex) {
+            // Continue to the next message
+        }
+
+        return true;
+    });
+
+    return foundSnsEvent;
+}
+
+module.exports.getSNSTrigger = getSNSTrigger;

src/triggers/aws_lambda.js

@@ -9,6 +9,7 @@ const serverlessEvent = require('../proto/event_pb.js');
 const errorCode = require('../proto/error_code_pb.js');
 const eventInterface = require('../event.js');
 const utils = require('../utils');
+const resourceUtils = require('../resource_utils/sqs_utils.js');
 
 /**
  * Fills the common fields for a trigger event
@@ -111,6 +112,10 @@ function createSQSTrigger(event, trigger) {
     }, {
         'Message Body': event.Records[0].body,
     });
+    const snsData = resourceUtils.getSNSTrigger(event.Records);
+    if (snsData != null) {
+        eventInterface.addToMetadata(trigger, { 'SNS Trigger': snsData });
+    }
 }
 
 /**