Merge branch 'master' of https://github.com/mitipi/serverless-iot-off…

…line

__tests__/sqlSyntax.test.js

@@ -1,7 +1,7 @@
 const {parseSelect} = require('../iotSql/parseSql')
 const {applySelect} = require('../iotSql/applySqlSelect')
 const {applyWhereClause} = require('../iotSql/applySqlWhere')
-const {topic, timestamp, clientid, accountid} = require('../iotSql/sqlFunctions')
+const {topic, timestamp, clientid, accountid, encode} = require('../iotSql/sqlFunctions')
 const {sqlParseTestData} = require('../testData')
 
 const log = () => {}
@@ -29,7 +29,9 @@ describe('SQL parser', () => {
         context: {
           topic: (index) => topic(index, parsed.topic),
           clientid: () => clientid(parsed.topic),
-          accountid: () => accountid()
+          timestamp: () => timestamp(),
+          accountid: () => accountid(),
+          encode: (field, encoding) => encode(payload, field, encoding)
         }
       })).toEqual(expected.event)
     })

iotSql/applySqlSelect.js

@@ -1,7 +1,6 @@
 const _ = require('lodash')
 const evalInContext = require('./eval')
 
-const BASE64_PLACEHOLDER = '*b64'
 const brace = new Buffer('{')[0]
 const bracket = new Buffer('[')[0]
 const doubleQuote = new Buffer('"')[0]
@@ -25,11 +24,6 @@ const applySelect = ({select, payload, context}) => {
   let event = {}
   const json = maybeParseJSON(payload)
 
-  // if payload is Buffer initialize Buffer class from base64 string
-  const payloadReplacement = Buffer.isBuffer(payload)
-    ? `new Buffer('${payload.toString('base64')}', 'base64')`
-    : payload
-
   // iterate over select parsed array
   // ex. [{alias: 'serialNumber', field: 'topic(2)'}, {field: 'state.reported.preferences.*'}]
   for (let part of select) {
@@ -45,9 +39,8 @@ const applySelect = ({select, payload, context}) => {
       }
       // check if field is sqlFunction
     } else if (Object.keys(context).some((sqlFunc) => (new RegExp(`${sqlFunc}\\((.*)\\)`).test(field)))) {
-      let js = field.replace(BASE64_PLACEHOLDER, payloadReplacement)
       // execute sqlFunction
-      event[alias || field.replace(/\(()\)/, '')] = evalInContext(js, context)
+      event[alias || field.replace(/\(()\)/, '')] = evalInContext(field, context)
     } else {
       // event is some property on shadow
       let propPath = field.split('.')

iotSql/eval.js

@@ -1,20 +1,15 @@
-const evalInContext = (js, context) => {
-  const {clientid, topic, accountid, timestamp} = context
+const evalInContext = (expr, context) => {
+  let [func, fields] = expr.match(/(\w+)\((.*)\)/).slice(1, 3);
+  fields = fields
+    ? fields.split(",").map((f) => f.trim().replace(/['"]+/g, ""))
+    : [];
+
   try {
-    return eval(js)
+    return context[func](...fields);
   } catch (err) {
-    debugger
-    console.log(`failed to evaluate: ${js}`)
-    throw err
-  }
-}
-
-const encode = (data, encoding) => {
-  if (encoding !== 'base64') {
-    throw new Error('AWS Iot SQL encode() function only supports base64 as an encoding')
+    debugger;
+    console.log(`failed to evaluate: ${expr}`);
+    throw err;
   }
-
-  return data.toString(encoding)
-}
-
-module.exports = evalInContext
+};
+module.exports = evalInContext;

iotSql/parseSql.js

@@ -1,19 +1,14 @@
-const BASE64_PLACEHOLDER = '*b64'
 const SQL_REGEX = /^SELECT (.*) FROM '([^']+)'/
 const SELECT_PART_REGEX = /^(.*?)(?: as (.*))?$/i
+const FIELDS_REGEX = /((\w+[\n\r\s]*\([^)]*\))|([^\n\r\s(,]+))([\n\r\s]+as[\n\r\s]+\w*)?/g
 const WHERE_REGEX = /WHERE (.*)/
 
 const parseSelect = sql => {
   const [select, topic] = sql.match(SQL_REGEX).slice(1)
   const [whereClause] = (sql.match(WHERE_REGEX) || []).slice(1)
 
   return {
-    select: select
-    // hack
-      .replace("encode(*, 'base64')", BASE64_PLACEHOLDER)
-      .split(',')
-      .map(s => s.trim())
-      .map(parseSelectPart),
+    select: select.match(FIELDS_REGEX).map(parseSelectPart),
     topic,
     whereClause
   }

iotSql/sqlFunctions.js

@@ -1,3 +1,5 @@
+const _ = require('lodash')
+
 module.exports = {
   topic: (index, topicUrl) => (typeof index !== 'undefined') ? topicUrl.split('/')[(index - 1)] : topicUrl,
   clientid: (topicUrl) => {
@@ -8,5 +10,30 @@ module.exports = {
     }
   },
   timestamp: () => (new Date()).getTime(),
-  accountid: () => process.env.AWS_ACCOUNT_ID
+  accountid: () => process.env.AWS_ACCOUNT_ID,
+  encode: (message, field, encoding) => {
+    if (encoding !== "base64") {
+      throw new Error(
+        "AWS Iot SQL encode() function only supports base64 as an encoding"
+      );
+    }
+    if (field === "*") {
+      return Buffer.from(message).toString("base64");
+    }
+
+    let payload;
+    try {
+      payload = JSON.parse(message);
+    } catch (e) {
+      console.log(e);
+    }
+
+    const value = _.get(payload, field);
+    if (!value) {
+      throw new Error(
+        `Failed to evaluate encode(${field}, 'base64'): Cannot find ${field} in payload`
+      );
+    }
+    return Buffer.from(value.toString()).toString("base64");
+  },
 }

ruleHandler.js

@@ -9,7 +9,7 @@ const { fillSubstitutionTemplates } = require('./iotSql/substitutionTemplates')
 const mqtt = require('mqtt')
 const mqttMatch = require('mqtt-match')
 const _ = require('lodash')
-const {topic, accountid, clientid, timestamp} = require('./iotSql/sqlFunctions')
+const {topic, accountid, clientid, timestamp, encode} = require('./iotSql/sqlFunctions')
 
 /**
  * Searches serverless.yml for functions configurations.
@@ -172,7 +172,8 @@ module.exports = (slsOptions, slsService, serverless, log) => {
               topic: (index) => topic(index, topicUrl),
               clientid: () => clientid(topicUrl),
               timestamp: () => timestamp(),
-              accountid: () => accountid()
+              accountid: () => accountid(),
+              encode: (field, encoding) => encode(message, field, encoding)
             }
           })
 

testData.js

@@ -85,5 +85,39 @@ module.exports.sqlParseTestData = [
         clientid: 'test_client'
       }
     }
+  },
+  {
+    sql: `SELECT encode(*, 'base64') as encodedPayload FROM '$aws/things/sn:123/shadow/update'`,
+    payload: `{"state": {"reported": {"mode": "STAND_BY"}}}`,
+    expected: {
+      parsed: {
+        select: [{ field: "encode(*, 'base64')", alias: "encodedPayload" }],
+        topic: "$aws/things/sn:123/shadow/update",
+      },
+      whereEvaluatesTo: true,
+      event: {
+        encodedPayload:
+          "eyJzdGF0ZSI6IHsicmVwb3J0ZWQiOiB7Im1vZGUiOiAiU1RBTkRfQlkifX19",
+      },
+    },
+  },
+  {
+    sql: `SELECT encode(state.reported.mode, 'base64') as encodedPayload FROM '$aws/things/sn:123/shadow/update'`,
+    payload: `{"state": {"reported": {"mode": "STAND_BY"}}}`,
+    expected: {
+      parsed: {
+        select: [
+          {
+            field: "encode(state.reported.mode, 'base64')",
+            alias: "encodedPayload",
+          },
+        ],
+        topic: "$aws/things/sn:123/shadow/update",
+      },
+      whereEvaluatesTo: true,
+      event: {
+        encodedPayload: "U1RBTkRfQlk=",
+      },
+    },
   }
 ]