timer_property

Author: FranckTala

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 1.2.2 (August 25, 2021)
+* Added delay timer property in re-assemble action.
+
 ## 1.2.1 (July 23, 2021)
 * Implemented support of maester storage in `Re-assembled message` action (maester-client library 3.3.0)
 

README.md

@@ -112,16 +112,19 @@ If at any point there is more than a 15 second gap in messages, then the group w
 * 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.
+* 
 
 #### 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.
 
-```messageData``` - object for providing some data derived from the steps between splitting and re-assembling
+```messageData``` - object for providing some data derived from the steps between splitting and re-assembling.
+
+```groupSize``` - Number of messages in the group.
+
+```Delay timer(in ms)``` - Time the process waits when no incoming messages before emiting(Max 40000 miliseconds)
 
 ## Known limitations (common for the component)
 No.

component.json

@@ -50,24 +50,37 @@
           "properties": {
             "groupSize": {
               "type": "number",
-              "required": true,
-              "title": "Number of messages produced by splitter"
+              "required": false,
+              "title": "Number of messages produced by splitter",
+              "order": 3
             },
             "groupId": {
               "type": "string",
               "required": true,
-              "title": "Unique ID to describe the group"
+              "title": "Unique ID to describe the group",
+              "order": 5
             },
             "messageId": {
               "type": "string",
               "required": true,
-              "title": "Unique ID to describe this message"
+              "title": "Unique ID to describe this message",
+              "order": 4
             },
             "messageData": {
               "title": "Message Data",
               "required": false,
               "type": "object",
-              "properties": {}
+              "properties": {},
+              "order": 1
+            },
+            "timersec": {
+              "type": "number",
+              "required": false,
+              "help":{
+                "description": "Time the process waits when no incoming messages before emiting(Default 20000 miliseconds)"
+              },
+              "title": "Delay timer(in ms)",
+              "order": 2
             }
           }
         }

lib/actions/reassemble.js

@@ -2,6 +2,16 @@
 const { messages } = require('elasticio-node');
 const ObjectStorageWrapperExtended = require('./utils-wrapper/ObjectStorageWrapperExtended');
 
+let timeHandle;
+var groupList = [];
+var groupElements = [{groupSize: undefined, groupId: undefined, messageData:undefined}];
+
+async function timer(this_) {
+  for(var i = 1; i < groupElements.length; i++){
+    await this_.emit('data', messages.newMessageWithBody( groupElements[i] ))  
+  }
+}
+
 async function processAction(msg) {
   const storage = new ObjectStorageWrapperExtended(this);
   const {
@@ -17,13 +27,18 @@ async function processAction(msg) {
     messageData,
   };
 
-  if (groupSize <= 0) {
-    throw new Error('Size must be a positive integer.');
-  }
+  // in case no delay was defined  
+  var delay = msg.body.timersec ?  msg.body.timersec : '40000'
+  if(delay >= 40000){delay == 40000}
+  clearTimeout(timeHandle);
+  timeHandle = setTimeout(timer, delay, this); 
+
   if (!messageData) {
     incomingData[messageId] = undefined;
   }
-
+  if (groupSize <= 0) {
+    throw new Error('Size must be a positive integer.');
+  }
   const {
     messageGroup,
     messageGroupId,
@@ -53,19 +68,73 @@ async function processAction(msg) {
     `Saw message ${messageId} of group ${groupId}. 
     Currently the group has ${messagesNumberSeen} of ${messageGroupSize} message(s).`,
   );
+  if(messageGroupSize !== '' && messageGroupSize !== undefined ){
+    clearTimeout(timeHandle);
+
+    if(groupList.includes(groupId)){
+      parsedMessageGroup.messages.forEach((message) => {
+        incomingData[message.messageId] = message.messageData;
+      });
+      
+      for(var key in groupElements){  
+        if(groupElements[key].groupId === groupId){
+          groupElements[key].groupSize == messagesNumberSeen
+          groupElements[key].messageData == incomingData
+        }
+      }
+    }
+    else{
+      parsedMessageGroup.messages.forEach((message) => {
+        incomingData[message.messageId] = message.messageData;
+      });
 
-  if (messagesNumberSeen >= messageGroupSize) {
-    parsedMessageGroup.messages.forEach((message) => {
-      incomingData[message.messageId] = message.messageData;
-    });
+      groupList.push(groupId)
+      groupElements.push({groupSize: messagesNumberSeen, groupId: groupId, messageData:incomingData})
+      
+    }
+    if (messagesNumberSeen >= messageGroupSize) {
+      parsedMessageGroup.messages.forEach((message) => {
+        incomingData[message.messageId] = message.messageData;
+      });
+  
+      await this.emit('data', messages.newMessageWithBody({
+        groupSize,
+        groupId,
+        messageData: incomingData,
+      }));
+      await storage.deleteObjectById(messageGroupId);
+      this.logger.info(`Message group with id ${messageGroupId} has been deleted`);
+      groupList = groupList.filter(def => def != groupId);
+      groupElements = groupElements.filter(def => def.groupId != groupId);
+    }
+    else{
+      timeHandle = setTimeout(timer, delay, this); 
+    }
+  }
+  else{
+    if(groupList.includes(groupId)){
+     
+      parsedMessageGroup.messages.forEach((message) => {
+        incomingData[message.messageId] = message.messageData;
+      });
+      
+      for(var key in groupElements){   
+        if(groupElements[key].groupId === groupId){
+          groupElements[key].groupSize == messagesNumberSeen
+          groupElements[key].messageData == incomingData
+        }
+      }
+    }
+    else{
+      
+      parsedMessageGroup.messages.forEach((message) => {
+        incomingData[message.messageId] = message.messageData;
+      });
 
-    await this.emit('data', messages.newMessageWithBody({
-      groupSize,
-      groupId,
-      messageData: incomingData,
-    }));
-    await storage.deleteObjectById(messageGroupId);
-    this.logger.info(`Message group with id ${messageGroupId} has been deleted`);
+      groupList.push(groupId)
+      groupElements.push({groupSize: messagesNumberSeen, groupId: groupId, messageData:incomingData})
+      
+    }
   }
 }