CUMULUS-350 Update queue-granules to use cumulus-message-adapter

cumulus/tasks/discover-granules/index.js

@@ -6,7 +6,7 @@ const log = require('@cumulus/common/log');
 
 /**
 * Discover granules
-* See schemas/input.json for detailed input schema
+* See schemas/input.json and schemas/config.json for detailed event description
 *
 * @param {Object} event - Lambda event object
 * @returns {Promise} - see schemas/output.json for detailed output schema

cumulus/tasks/queue-granules/index.js

@@ -1,52 +1,42 @@
 'use strict';
 
-const queueGranule = require('@cumulus/ingest/queue').queueGranule;
-const log = require('@cumulus/common/log');
+const cumulusMessageAdapter = require('@cumulus/cumulus-message-adapter-js');
+const { enqueueGranuleIngestMessage } = require('@cumulus/ingest/queue');
 
 /**
-* Callback function provided by aws lambda. See https://docs.aws.amazon.com/lambda/latest/dg/nodejs-prog-model-handler.html#nodejs-prog-model-handler-callback
-* @callback lambdaCallback
-* @param {object} error
-* @param {object} output - output object matching schemas/output.json
-* @param {integer} output.granules_queued
-*/
-
-/**
-* For each Granule, generate a new SF messages send to the step function queue to be executed
-* @param  {object} event lambda event object
-* @param  {object} event.input
-* @param  {array} event.input.granules
-* @param  {object} context Lambda context object. See https://docs.aws.amazon.com/lambda/latest/dg/nodejs-prog-model-context.html
-* @param  {lambdaCallback} callback callback function
-* @return {undefined}
+* See schemas/input.json and schemas/config.json for detailed event description
+*
+* @param {Object} event - Lambda event object
+* @returns {Promise} - see schemas/output.json for detailed output schema
+*   that is passed to the next task in the workflow
 **/
-function handler(event, context, cb) {
-  const config = event.config;
-  const stack = config.stack;
-  const bucket = config.bucket;
-  const queueUrl = config.queueUrl;
-  const templateUri = config.templateUri;
-  const provider = config.provider;
-  const collection = config.collection;
+async function queueGranules(event) {
   const granules = event.input.granules || [];
 
-  const queuedGranules = granules.map(g => queueGranule(
-    g,
-    queueUrl,
-    templateUri,
-    provider,
-    collection,
-    null,
-    stack,
-    bucket
-  ));
+  await Promise.all(
+    granules.map((granule) => enqueueGranuleIngestMessage(
+      granule,
+      event.config.queueUrl,
+      event.config.granuleIngestMessageTemplateUri,
+      event.config.provider,
+      event.config.collection,
+      event.input.pdr
+    ))
+  );
 
-  return Promise.all(queuedGranules).then(() => {
-    cb(null, { granules_queued: queuedGranules.length });
-  }).catch((e) => {
-    log.error(e);
-    cb(e);
-  });
+  return { granules_queued: granules.length };
 }
+exports.queueGranules = queueGranules;
 
-module.exports.handler = handler;
+/**
+ * Lambda handler
+ *
+ * @param {Object} event - a Cumulus Message
+ * @param {Object} context - an AWS Lambda context
+ * @param {Function} callback - an AWS Lambda handler
+ * @returns {undefined} - does not return a value
+ */
+function handler(event, context, callback) {
+  cumulusMessageAdapter.runCumulusTask(queueGranules, event, context, callback);
+}
+exports.handler = handler;

cumulus/tasks/queue-granules/package.json

@@ -39,6 +39,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@cumulus/common": "^1.0.1",
+    "@cumulus/cumulus-message-adapter-js": "0.0.1-beta.3",
     "@cumulus/ingest": "^1.0.1",
     "babel-core": "^6.25.0",
     "babel-loader": "^6.2.4",
@@ -51,6 +52,7 @@
   },
   "devDependencies": {
     "@mapbox/mock-aws-sdk-js": "0.0.5",
-    "ava": "^0.21.0"
+    "ava": "^0.21.0",
+    "lodash": "^4.17.5"
   }
 }

cumulus/tasks/queue-granules/schemas/config.json

@@ -0,0 +1,18 @@
+{
+  "title": "QueueGranulesConfig",
+  "description": "Describes the config used by the queue-granules task",
+  "type": "object",
+  "required": [
+    "collection",
+    "provider",
+    "queueUrl",
+    "granuleIngestMessageTemplateUri"
+  ],
+  "additionalProperties": false,
+  "properties": {
+    "collection": { "type": "object" },
+    "provider": { "type": "object" },
+    "queueUrl": { "type": "string" },
+    "granuleIngestMessageTemplateUri": { "type": "string" }
+  }
+}

cumulus/tasks/queue-granules/schemas/input.json.txt → cumulus/tasks/queue-granules/schemas/input.json

@@ -2,16 +2,24 @@
   "title": "QueueGranulesInput",
   "description": "Describes the input and config used by the queue-granules task",
   "type": "object",
+  "require": [ "granules" ],
   "properties": {
+    "pdr": {
+      "type": "object",
+      "required": ["name", "path"],
+      "properties": {
+        "name": { "type": "string" },
+        "path": { "type": "string" }
+      }
+    },
     "granules": {
       "type": "array",
       "items": {
         "type": "object",
+        "required": [ "files", "granuleId" ],
         "properties": {
-          "name": { "type": "string" },
           "granuleId": { "type": "string" },
-          "bucket": { "type": "string" },
-          "url_path": { "type": "string" }
+          "files": { "type": "array" }
         }
       }
     }

cumulus/tasks/queue-granules/schemas/output.json.txt → cumulus/tasks/queue-granules/schemas/output.json

@@ -2,9 +2,8 @@
   "title": "QueueGranulesOutput",
   "description": "Describes the output produced by the queue-granules task",
   "type": "object",
+  "required": [ "granules_queued" ],
   "properties": {
-    "granules_queued": {
-      "type": "integer"
-    }
+    "granules_queued": { "type": "integer" }
   }
 }

cumulus/tasks/queue-granules/tests/.eslintrc.json

@@ -0,0 +1,5 @@
+{
+  "rules": {
+    "no-param-reassign": "off"
+  }
+}