Reduce max delay to 20sec (#91)

Author: EmptyInfinity

.circleci/config.yml

@@ -72,14 +72,14 @@ commands:
 jobs:
   test:
     docker:
-      - image: circleci/node:14-stretch
+      - image: circleci/node:16-stretch
     steps:
       - checkout
       - node/install:
           node-version: << pipeline.parameters.node-version >>
       - run:
           name: Audit Dependencies
-          command: npm audit --audit-level=high
+          command: npm audit --production --audit-level=high
       - node/install-packages:
           cache-path: ./node_modules
           override-ci-command: npm install
@@ -88,7 +88,7 @@ jobs:
           command: npm test
   build:
     docker:
-      - image: circleci/node:14-stretch
+      - image: circleci/node:16-stretch
         user: root
     steps:
       - checkout

CHANGELOG.md

@@ -1,3 +1,10 @@
+## 1.4.2 (October 07, 2022)
+* Update Sailor version to 2.7.0
+* Update @elastic.io/maester-client version to 4.0.3
+* Get rid of vulnerabilities in dependencies
+* Fix message processing
+* Maximum `Delay timer` in `Re-assembled message` action reduced to 20 sec
+
 ## 1.4.1 (April 08, 2022)
 * Update Sailor version to 2.6.27
 * Get rid of vulnerabilities in dependencies

README.md

@@ -114,7 +114,7 @@ If all the messages in the group do not arrive, then the group will not be emitt
 * The group is dropped if there are any unexpected restarts to the container.
 * In case only a groupSize is given and no delay timer is specified. The size of the group must be known by all group members.
 * In case of using the delay timer. Messages are only emitted when all parts arrive. Emitting a message only when the first part arrives isn't supported.
-* The delay timer can not exceed 40,000 milliseconds. If more than this maximum is given, then this maximum will be used instead.
+* The delay timer can not exceed 20,000 milliseconds. If more than this maximum is given, then this maximum will be used instead.
 
 #### List of Expected Config fields
 ```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.

component.json

@@ -1,7 +1,7 @@
 {
   "title": "Splitter",
-  "version": "1.4.1",
   "description": "Splits a message into multiple messages.",
+  "version": "1.4.2",
   "actions": {
     "split": {
       "deprecated": true,
@@ -57,4 +57,4 @@
       }
     }
   }
-}
+}

lib/actions/reassemble.js

@@ -11,7 +11,7 @@ async function timer(this_) {
   for (let i = 0; i < groupList.length; i += 1) {
     const storage = new ObjectStorageWrapperExtended(this_);
     // eslint-disable-next-line no-await-in-loop
-    const results = await storage.lookupParsedObjectById(groupList[i]);
+    const results = await storage.lookupObjectById(groupList[i]);
     const incomingData = {};
     results.messages.forEach((message) => {
       incomingData[message.messageId] = message.messageData;
@@ -37,6 +37,7 @@ async function processAction(msg, cfg) {
     groupId,
     messageId = uuidv4(),
     messageData,
+    timersec = 20000,
   } = msg.body;
   const incomingData = {};
   const object = {
@@ -54,7 +55,7 @@ async function processAction(msg, cfg) {
     }
   }
   if (mode === 'timeout') {
-    if (msg.body.timersec <= 0) {
+    if (timersec <= 0) {
       throw new Error('Delay timer must be a positive integer.');
     }
   }
@@ -74,21 +75,19 @@ async function processAction(msg, cfg) {
     this.logger.info('Existed Group found. Added message');
     this.logger.info(`Saved messages: ${Object.keys(messageGroup.messageIdsSeen).join(', ')}`);
   }
-
-  const parsedMessageGroup = await storage.lookupParsedObjectById(messageGroupId);
+  const parsedMessageGroup = await storage.lookupObjectById(messageGroupId);
   const filteredMessages = parsedMessageGroup.messages
     .filter((message) => message.messageId !== messageId);
   filteredMessages.push(object);
   parsedMessageGroup.messages = filteredMessages;
-  await storage.updateObject(messageGroupId, parsedMessageGroup);
+  await storage.updateObjectById(messageGroupId, parsedMessageGroup);
   const messagesNumberSeen = Object.keys(parsedMessageGroup.messageIdsSeen).length;
 
   this.logger.info(
     `Saw message ${messageId} of group ${groupId}. 
     Currently the group has ${messagesNumberSeen} of ${messageGroupSize} message(s).`,
   );
-
-  // when grouSized option is selected
+  // when groupSized option is selected
   if (mode === 'groupSize') {
     if (messagesNumberSeen >= messageGroupSize) {
       parsedMessageGroup.messages.forEach((message) => {
@@ -106,7 +105,7 @@ async function processAction(msg, cfg) {
 
   // When delay timer option is selected
   if (mode === 'timeout') {
-    delay = (msg.body.timersec >= 40000) ? 40000 : msg.body.timersec;
+    delay = (timersec >= 20000) ? 20000 : timersec;
     clearTimeout(timeHandle);
     timeHandle = setTimeout(timer, delay, this);
     if (!groupList.includes(messageGroupId)) {
@@ -116,7 +115,7 @@ async function processAction(msg, cfg) {
 
   // When both groupSize and delay timer option is selected
   if (mode === 'groupSize&timeout') {
-    delay = (msg.body.timersec >= 40000) ? 40000 : msg.body.timersec;
+    delay = (timersec >= 20000) ? 20000 : timersec;
     clearTimeout(timeHandle);
     timeHandle = setTimeout(timer, delay, this);
 
@@ -142,7 +141,7 @@ async function processAction(msg, cfg) {
 }
 
 //-------------------------------------------------------------------------------------
-// Dynamic drop-down logic starts hier
+// Dynamic drop-down logic starts here
 
 async function getMetaModel(cfg) {
   if (cfg.mode === 'groupSize') {

lib/actions/utils-wrapper/ObjectStorageWrapperExtended.js

@@ -1,4 +1,4 @@
-const { ObjectStorageWrapper } = require('@elastic.io/maester-client/dist/ObjectStorageWrapper');
+const { ObjectStorageWrapper } = require('@elastic.io/maester-client');
 
 class ObjectStorageWrapperExtended extends ObjectStorageWrapper {
   constructor(context) {
@@ -8,11 +8,6 @@ class ObjectStorageWrapperExtended extends ObjectStorageWrapper {
     this.TTL_TWO_DAYS = 172800;
   }
 
-  async lookupParsedObjectById(messageGroupId) {
-    const messageGroup = await this.lookupObjectById(messageGroupId);
-    return JSON.parse(messageGroup);
-  }
-
   async createMessageGroupIfNotExists(externalId, messageGroupSize) {
     this.logger.info('Processing creation of the new message group');
     const messageGroups = await this.lookupObjectsByQueryParameters(
@@ -27,7 +22,7 @@ class ObjectStorageWrapperExtended extends ObjectStorageWrapper {
         messages: [],
         messageIdsSeen: {},
       };
-      const { objectId: messageGroupId } = await this.createObject(
+      const messageGroupId = await this.createObject(
         newMessageGroup, [{ key: this.EXTERNAL_ID_QUERY_HEADER_NAME, value: externalId }],
         [], this.TTL_TWO_DAYS,
       );
@@ -38,19 +33,19 @@ class ObjectStorageWrapperExtended extends ObjectStorageWrapper {
     }
     this.logger.info('MessageGroup found');
     const messageGroupId = messageGroups[0].objectId;
-    const parsedMessageGroup = await this.lookupParsedObjectById(messageGroupId);
+    const messageGroup = await this.lookupObjectById(messageGroupId);
     return {
-      messageGroup: parsedMessageGroup, messageGroupSize, messageGroupId, isCreated: false,
+      messageGroup, messageGroupSize, messageGroupId, isCreated: false,
     };
   }
 
   async createNewObjectInMessageGroup(object, messageGroupId) {
     this.logger.info('Processing creation of the new object');
-    const parsedMessageGroup = await this.lookupParsedObjectById(messageGroupId);
+    const messageGroup = await this.lookupObjectById(messageGroupId);
     this.logger.info('...Updating message group');
-    parsedMessageGroup.messageIdsSeen[object.messageId] = object.messageId;
-    return this.updateObject(messageGroupId, {
-      ...parsedMessageGroup, messages: [...parsedMessageGroup.messages, object],
+    messageGroup.messageIdsSeen[object.messageId] = object.messageId;
+    return this.updateObjectById(messageGroupId, {
+      ...messageGroup, messages: [...messageGroup.messages, object],
     });
   }
 }