Add Re-assemble (#42)

* Apply AirBnB Style Rules.
* 1.1.0
* Update dependencies.

Author: jhorbulyk

.eslintrc.js

@@ -1,159 +1,7 @@
-'use strict';
-
-const ERROR = 'error';
-const WARN = 'warn';
-const ALWAYS = 'always';
-const NEVER = 'never';
-
 module.exports = {
-    'env': {
-        es6: true,
-        node: true,
-        jasmine: true,
-        mocha: true
-    },
-    "parserOptions": {
-        "ecmaVersion": 8
-    },
-    'extends': 'airbnb-base',
-    'rules': {
-        'indent': [
-            ERROR,
-            4,
-            {
-                SwitchCase: 1
-            }
-        ],
-        'linebreak-style': ERROR,
-        'quotes': [
-            WARN,
-            'single'
-        ],
-        'semi': [
-            ERROR,
-            ALWAYS
-        ],
-        'func-names': ERROR,
-        'no-empty': ERROR,
-        'no-empty-function': ERROR,
-        'brace-style': [
-            ERROR,
-            '1tbs',
-            { allowSingleLine: true }
-        ],
-        'no-multiple-empty-lines': ERROR,
-        'no-multi-spaces': ERROR,
-        'one-var': [
-            ERROR,
-            NEVER
-        ],
-        'quote-props': [
-            WARN,
-            'consistent-as-needed'
-        ],
-        'key-spacing': ERROR,
-        'space-unary-ops': [
-            ERROR,
-            {
-                words: true,
-                nonwords: false
-            }
-        ],
-        'no-spaced-func': ERROR,
-        'space-before-function-paren': [
-            ERROR,
-            {
-                anonymous: ALWAYS,
-                named: NEVER
-            }
-        ],
-        'arrow-body-style': [
-            WARN,
-            'as-needed'
-        ],
-        'array-bracket-spacing': ERROR,
-        'space-in-parens': ERROR,
-        'comma-dangle': ERROR,
-        'no-trailing-spaces': ERROR,
-        'yoda': ERROR,
-        'max-len': [
-            ERROR,
-            120
-        ],
-        'camelcase': [
-            ERROR,
-            {
-                properties: 'never'
-            }
-        ],
-        'new-cap': [
-            WARN,
-            {
-                capIsNewExceptions: ['Q']
-            }
-        ],
-        'comma-style': ERROR,
-        'curly': ERROR,
-        'object-curly-spacing': [
-            WARN,
-            ALWAYS
-        ],
-        'object-curly-newline':  [
-            ERROR,
-            {
-                ObjectExpression: {
-                    minProperties: 1
-                },
-                ObjectPattern: {
-                    minProperties: 5
-                }
-            }
-        ],
-        'object-property-newline': ERROR,
-        'template-curly-spacing': ERROR,
-        'dot-notation': ERROR,
-        'dot-location': [
-            ERROR,
-            'property'
-        ],
-        'func-style': [
-            ERROR,
-            'declaration',
-            {
-                allowArrowFunctions: true
-            }
-        ],
-        'eol-last': ERROR,
-        'space-infix-ops': ERROR,
-        'keyword-spacing': ERROR,
-        'space-before-blocks': ERROR,
-        'no-invalid-this': ERROR,
-        'consistent-this': ERROR,
-        'no-this-before-super': ERROR,
-        'no-unreachable': ERROR,
-        'no-sparse-arrays': ERROR,
-        'array-callback-return': ERROR,
-        'strict': [
-            WARN,
-            'global'
-        ],
-        'eqeqeq': ERROR,
-        'no-use-before-define': WARN,
-        'no-undef': ERROR,
-        'no-unused-vars': WARN,
-        'no-mixed-spaces-and-tabs': ERROR,
-        'operator-linebreak': [
-            ERROR,
-            'before'
-        ],
-        'no-console': [
-            WARN,
-            {
-                'allow': [
-                    'warn',
-                    'error'
-                ]
-            }
-        ]
-    }
-};
+  'extends': 'airbnb-base',
+  'env': {
+    'mocha': true,
+    'node': true,
+  }
+};

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.1.0 (May 7, 2020)
+* Add re-assemble action
+* Update dependencies
+
 ## 1.0.6 (March 26, 2020)
 * Add support for new getFlowVariables and handlePathrough jsonata functions
 

README.md

@@ -94,6 +94,37 @@ and the JSONata expression `Phone.{type: number}`, an object constructor, the ac
 #### List of Expected Config fields
 ```Split Property``` - use this field to choose a separator.
 
+### Re-assemble Messages 
+**(Beta)**
+
+Inverse of the split action: Given a stream of incoming messages that which have 
+been split apart by a split action (or similar), produce one message once all 
+message parts have arrived.
+
+Supported:
+* Messages can be re-ordered in the flow
+* If messages are re-delivered as a result of the platform's at once delivery guarantee, does not trigger false positives
+* Messages from one original message can be interleaved with messages from another original message
+(e.g. Two overlapping webhook calls arrive and the flow has components where parallel processing > 1.)
+
+Limitations:
+* All groups must have one or more messages. (i.e. No groups of size 0). 
+Can't do re-grouping when a split is done on an empty array. (i.e. No empty for each pattern supported)
+* If all messages in a group fail to arrive at the re-assemble action (because one message suffered an error earlier in the flow)
+then this component will silently discard the group.
+* All messages must arrive within the same container lifetime.  
+If at any point there is more than a 15 second gap in messages, then the group will be silently discarded.  
+* The group is dropped if there are any unexpected restarts to the container.
+* Size of the group must be known by all group members.
+* Messages are only emitter when all parts arrive. Emitting a message only when the first part arrives isn't supported.
+* The contents of data that are picked up by the sub-messages aren't passed forward to future steps.
+
+#### List of Expected Config fields
+```groupSize``` - Number of messages in the group
+```groupId``` - Globally unique id for the group to distinguish it from other groups. This value needs to be the same for all messages in a group.
+```messageId``` - Id for a message to distinguish it from other messages in the group. 
+Must be unique per group but does not have to be globally unique. This value needs to be different for all messages in a group.
+
 ## Known limitations (common for the component)
 No.
 

component.json

@@ -27,6 +27,38 @@
           "viewClass": "JSONataView"
         }
       }
+    },
+    "reassemble": {
+      "title": "Re-assemble Messages",
+      "help": {
+        "description": "Inverse operation from split: Re-assemble a group of messages that have previously been split",
+        "link": "#re-assemble-messages"
+      },
+      "main": "./lib/actions/reassemble.js",
+      "fields": {},
+      "metadata": {
+        "in": {
+          "type": "object",
+          "required": true,
+          "properties": {
+            "groupSize": {
+              "type": "number",
+              "required": true,
+              "title": "Number of messages produced by splitter"
+            },
+            "groupId": {
+              "type": "string",
+              "required": true,
+              "title": "Unique ID to describe the group"
+            },
+            "messageId": {
+              "type": "string",
+              "required": true,
+              "title": "Unique ID to describe this message"
+            }
+          }
+        }
+      }
     }
   }
 }

lib/actions/reassemble.js

@@ -0,0 +1,39 @@
+const { messages } = require('elasticio-node');
+
+const groupsSeen = {};
+
+async function processAction(msg) {
+  const {
+    groupSize,
+    groupId,
+    messageId,
+  } = msg.body;
+
+  if (groupSize <= 0) {
+    throw new Error('Size must be a positive integer.');
+  }
+
+  if (!groupsSeen[groupId]) {
+    groupsSeen[groupId] = {
+      groupSize,
+      messageIdsSeen: new Set(),
+    };
+  }
+
+  groupsSeen[groupId].messageIdsSeen.add(messageId);
+  const numberSeen = groupsSeen[groupId].messageIdsSeen.size;
+
+  this.logger.info(
+    `Saw message ${messageId} of group ${groupId} Currently the group has ${numberSeen} of ${groupSize} message(s).`,
+  );
+
+  if (numberSeen >= groupSize) {
+    await this.emit('data', messages.newMessageWithBody({
+      groupSize,
+      groupId,
+    }));
+    delete groupsSeen[groupId];
+  }
+}
+
+exports.process = processAction;

lib/actions/splitOnJsonata.js

@@ -6,26 +6,26 @@ const { messages } = require('elasticio-node');
  * @this processAction
  */
 async function processAction(msg, cfg) {
-    const split = JsonataTransform.jsonataTransform(msg, cfg, this);
+  const split = JsonataTransform.jsonataTransform(msg, cfg, this);
 
-    if (!Array.isArray(split)) {
-        await this.emit('error', new Error('The evaluated expression must be an array.'));
-        return;
-    }
+  if (!Array.isArray(split)) {
+    await this.emit('error', new Error('The evaluated expression must be an array.'));
+    return;
+  }
 
-    if (_.find(split, (elem) => !_.isObject(elem))) {
-        await this.emit('error', new Error('Splitting arrays of objects only!'));
-        return;
-    }
+  if (_.find(split, (elem) => !_.isObject(elem))) {
+    await this.emit('error', new Error('Splitting arrays of objects only!'));
+    return;
+  }
 
-    this.logger.info('Splitting the incoming message into %s messages', split.length);
+  this.logger.info('Splitting the incoming message into %s messages', split.length);
 
-    for (let i = 0; i < split.length; i += 1) {
-        if (split[i]) {
-            // eslint-disable-next-line no-await-in-loop
-            await this.emit('data', messages.newMessageWithBody(split[i]));
-        }
+  for (let i = 0; i < split.length; i += 1) {
+    if (split[i]) {
+      // eslint-disable-next-line no-await-in-loop
+      await this.emit('data', messages.newMessageWithBody(split[i]));
     }
+  }
 }
 
 exports.process = processAction;