Merge pull request #77 from MitocGroup/sync_vanilla

🍔 Sync backend with vanilla ms

Author: CCristi

src/deep-root-angular/backend/resources.json

@@ -21,6 +21,15 @@
     }
   },
   "ddb-eventual-consistency": {
+    "listen-queues": {
+      "description": "Listen for SQS offload queues messages",
+      "type": "lambda",
+      "forceUserIdentity": false,
+      "scope": "private",
+      "cron": "0/1 * * * ? *",
+      "methods": ["GET"],
+      "source": "src/ddb-eventual-consistency/listen-queues"
+    },
     "pool-queue": {
       "description": "Pool the SQS queue for DynamoDB offloaded operations on alarm received from CloudWatchAlarm through an SNS topic",
       "type": "lambda",

src/deep-root-angular/backend/src/async-config/dump/Handler.es6

@@ -98,7 +98,7 @@ export default class extends DeepFramework.Core.AWS.Lambda.Runtime {
    */
   _selfDisable(cb) {
     let resource = this.kernel.get('resource');
-    let lambda = resource.get('@deep-root-angular:scheduler:rule');
+    let lambda = resource.get(this._selfDisableResourceId);
     let payload = {
       effect: 'disable',
       lambdaName: this.context.functionName
@@ -107,6 +107,13 @@ export default class extends DeepFramework.Core.AWS.Lambda.Runtime {
     lambda.request(payload).useDirectCall().send(cb);
   }
 
+  /**
+   * @returns {String}
+   */
+  get _selfDisableResourceId() {
+    return `@${this.kernel.microservice().identifier}:scheduler:rule`;
+  }
+
   /**
    * @param {Function} callback
    * @private

src/deep-root-angular/backend/src/async-config/dump/package.json

@@ -10,7 +10,7 @@
     }
   ],
   "bugs": {
-    "url": "https://github.com/MitocGroup/deep-microservices-root-angularjs/issues"
+    "url": "https://github.com/MitocGroup/deep-microservices-root-angular/issues"
   },
   "keywords": [
     "DEEP",

src/deep-root-angular/backend/src/ddb-eventual-consistency/listen-queues/Handler.es6

@@ -0,0 +1,125 @@
+/**
+ * Created by acucer on 6/14/16.
+ */
+
+'use strict';
+
+import AWS from 'aws-sdk';
+import DeepFramework from 'deep-framework';
+
+export default class extends DeepFramework.Core.AWS.Lambda.Runtime {
+  /**
+   * @param {*} args
+   */
+  constructor(...args) {
+    super(...args);
+  }
+
+  /**
+   * @param {*} request
+   */
+  handle(request) {
+    Promise
+      .all(this._queues.map(queueConfig => this._checkQueue(queueConfig)))
+      .then(this.createResponse({}).send);
+  }
+
+  /**
+   * @param {Object} queueConfig
+   * @returns {Promise|*}
+   */
+  _checkQueue(queueConfig) {
+    return new Promise(resolve => {
+      let payload = {
+        QueueUrl: queueConfig.url,
+        AttributeNames: ['ApproximateNumberOfMessages',],
+      };
+
+      queueConfig.sqs.getQueueAttributes(payload, (error, data) => {
+        if (error) {
+          console.error(error);
+          return resolve();
+        }
+
+        let attrs = data.Attributes || {
+          ApproximateNumberOfMessages: '0', // o_O AWS sends a string
+        };
+
+        if (parseInt(attrs.ApproximateNumberOfMessages) > 0) {
+          return this._invokePullQueue(queueConfig.name)
+            .then(resolve)
+            .catch(error => {
+              console.error(error);
+              resolve();
+            });
+        }
+
+        resolve();
+      });
+    });
+  }
+
+  /**
+   * @param {String} queueName
+   * @returns {Promise|*}
+   */
+  _invokePullQueue(queueName) {
+    return new Promise((resolve, reject) => {
+      let resource = this.kernel.get('resource');
+
+      resource
+        .get(this._pullQueueResourceId)
+        .request({queueName,})
+        .invokeAsync()
+        .send((response) => {
+          response.isError ? reject(response.error) : resolve();
+        });
+    });
+  }
+
+  /**
+   * @returns {String}
+   */
+  get _pullQueueResourceId() {
+    return `@${this.kernel.microservice().identifier}:ddb-eventual-consistency:pool-queue`;
+  }
+
+  /**
+   * @param {String} queueUrl
+   * @returns {String}
+   * @private
+   */
+  _getRegionFromSqsQueueUrl(queueUrl) {
+    let regionParts = queueUrl.match(/\.([^\.]+)\.amazonaws\.com\/.*/i);
+
+    if (!regionParts || regionParts.length === 0) {
+      throw new Error(queueUrl, 'Unable to extract AWS region.');
+    }
+
+    return regionParts[1];
+  }
+
+  /**
+   * @returns {Object[]}
+   */
+  get _queues() {
+    let result = [];
+    let queues = this.kernel.config.dbOffloadQueues;
+
+    for (let modelName in queues) {
+      if (!queues.hasOwnProperty(modelName)) {
+        continue;
+      }
+
+      let queueConfig = queues[modelName];
+      let name = queueConfig.name;
+      let url = queueConfig.url;
+      let region = this._getRegionFromSqsQueueUrl(url);
+      let sqs = new AWS.SQS({region,});
+
+      result.push({name, url, region, sqs,});
+    }
+
+    return result;
+  }
+}

src/deep-root-angular/backend/src/ddb-eventual-consistency/listen-queues/bootstrap.es6

@@ -0,0 +1,10 @@
+/**
+ * Created by acucer on 3/17/16.
+ */
+
+'use strict';
+
+import DeepFramework from 'deep-framework';
+import Handler from './Handler';
+
+export default DeepFramework.LambdaHandler(Handler);

src/deep-root-angular/backend/src/ddb-eventual-consistency/listen-queues/package.json

@@ -0,0 +1,30 @@
+{
+  "name": "deep-system-ddb-eventual-consistency-listen-queues",
+  "version": "0.0.1",
+  "description": "Listen for SQS offload queues messages",
+  "author": "Mitoc Group <hello@mitocgroup.com>",
+  "contributors": [
+    {
+      "name": "Alexander Cucer",
+      "email": "acucer@mitocgroup.com"
+    }
+  ],
+  "bugs": {
+    "url": "https://github.com/MitocGroup/deep-microservices-root-angular/issues"
+  },
+  "keywords": [
+    "DEEP",
+    "Config",
+    "SQS",
+    "EventualConsistency",
+    "DynamoDB"
+  ],
+  "scripts": {
+    "postinstall": "npm run compile",
+    "compile": "deepify compile es6 `pwd`"
+  },
+  "dependencies": {
+    "deep-framework": "^1.x.x"
+  },
+  "private": true
+}

src/deep-root-angular/backend/src/ddb-eventual-consistency/pool-queue/Handler.es6

@@ -18,6 +18,7 @@ export default class extends DeepFramework.Core.AWS.Lambda.Runtime {
 
   /**
    * @param {*} request
+   * @todo refactor for direct call only (avoid SQS->SNS invokations)
    */
   handle(request) {
     let queueName = this._getQueueNameFromRequest(request);
@@ -28,6 +29,10 @@ export default class extends DeepFramework.Core.AWS.Lambda.Runtime {
     this._poolQueueUntilEmpty(queueUrl, sqs, dynamo).then(() => {
       let cloudWatch = this._cloudWatch(queueUrl);
 
+      if (!this._isInvokedBySns(request)) {
+        return this.createResponse({}).send();
+      }
+
       // @see deep-package-manager (queueName === alarmName)
       this._resetAlarmState(cloudWatch, queueName)
         .then(this.createResponse({}).send);
@@ -41,6 +46,7 @@ export default class extends DeepFramework.Core.AWS.Lambda.Runtime {
    * @param {AWS.CloudWatch|*} cloudWatch
    * @param {String} alarmName
    * @returns {Promise|*}
+   * @deprecated
    */
   _resetAlarmState(cloudWatch, alarmName) {
     return new Promise(resolve => {
@@ -220,13 +226,27 @@ export default class extends DeepFramework.Core.AWS.Lambda.Runtime {
     return regionParts[1];
   }
 
+  /**
+   * @param {*} request
+   * @returns {Boolean}
+   */
+  _isInvokedBySns(request) {
+    return !!request.getParam('Records');
+  }
+
   /**
    * @param {*} request
    * @returns {String}
    * @private
    * @todo: pre-validate event data
    */
   _getQueueNameFromRequest(request) {
+    if (!this._isInvokedBySns(request)) {
+      return request.getParam('queueName');
+    }
+
+    // @deprecated left for back compatibility only
+    // SQS event delivered through SNS topic
     return JSON.parse(request.getParam('Records')[0].Sns.Message).Trigger.Dimensions[0].value;
   }
 

src/deep-root-angular/backend/src/ddb-eventual-consistency/pool-queue/package.json

@@ -10,7 +10,7 @@
     }
   ],
   "bugs": {
-    "url": "https://github.com/MitocGroup/deep-microservices-root-angularjs/issues"
+    "url": "https://github.com/MitocGroup/deep-microservices-root-angular/issues"
   },
   "keywords": [
     "DEEP",

src/deep-root-angular/backend/src/scheduler/rule/package.json

@@ -10,7 +10,7 @@
     }
   ],
   "bugs": {
-    "url": "https://github.com/MitocGroup/deep-microservices-root-angularjs/issues"
+    "url": "https://github.com/MitocGroup/deep-microservices-root-angular/issues"
   },
   "keywords": [
     "DEEP",

src/deep-root-angular/parameters.json

@@ -16,14 +16,6 @@
       "default": "@deep-root-angular:img/favicon.ico",
       "example": "@deep-root-angular:img/favicon.ico"
     },
-    "storage|eventualConsistency|offloaderEndpoint": {
-      "displayName": "Eventual consistency offloading backend endpoint",
-      "type": "string",
-      "required": false,
-      "pattern": "^@[^:]+:[^:]+:[^:]+$",
-      "default": "@deep-root-angular:ddb-eventual-consistency:pool-queue",
-      "example": "@deep-root-angular:ddb-eventual-consistency:pool-queue"
-    },
     "pageLoader|src": {
       "displayName": "Page loader source",
       "type": "string",