fixing bugs

Author: FranckTala

README.md

@@ -93,9 +93,9 @@ and the JSONata expression `Phone.{type: number}`, an object constructor, the ac
 
 Inverse of the split action: Given a stream of incoming messages a sum message is generated.
 Has 3 different behaviour variants(options):
-* Use Group Size: A message is emitted once the group size is reached for the given group. If arriving messages for a particular group are less than the defined group size then the group is silently discarded.
-* Use Timeout: All incomming messages count towards the delay timer. Once no more message is received in this time frame there will be a emitted message for each group.
-* Use Group Size and Timeout: Specify both group size and delay timer. Groups that have reached their limit are emitted directly. Beyond that the action behaves as specifed in the line before.
+* Produce Groups of Fixed Size (Don't Emit Partial Groups): A message is emitted once the group size is reached for the given group. If arriving messages for a particular group are less than the defined group size then the group will not be emitted.
+* Group All Incoming Messages: All incomming messages will be gathered until there are no more incoming messages at which point messages will be emitted for each group.
+* Produce Groups of Fixed Size (Emit Partial Groups): Specify both group size and delay timer. Once a group is complete, that group will be emitted. Once there are no more incoming messages, then partially completed groups will also be emitted.
 
 Supported:
 * Messages can be re-ordered in the flow
@@ -106,8 +106,7 @@ Supported:
 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).
-* All messages must arrive within the same container lifetime.
-If all the messages in the group do not arrive, then the group will be silently discarded.
+If all the messages in the group do not arrive, then the group will not be emitted..
 * 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.

component.json

@@ -46,10 +46,10 @@
       "fields": {
         "mode": {
           "viewClass": "SelectView",
-          "label": "Select option",
+          "label": "Behavior",
           "required": true,
-          "model": {"groupSize": "Use Group Size", "timeout": "Use Timeout", "groupSize&timeout": "Use Group Size and Timeout"},
-          "prompt": "Select type"
+          "model": {"groupSize": "Produce Groups of Fixed Size (Don't Emit Partial Groups)", "timeout": "Group All Incoming Messages", "groupSize&timeout": "Produce Groups of Fixed Size (Emit Partial Groups)"},
+          "prompt": "Select behavior"
         }
       }
     }

lib/actions/reassemble.js

@@ -4,12 +4,27 @@ const ObjectStorageWrapperExtended = require('./utils-wrapper/ObjectStorageWrapp
 
 let timeHandle;
 let groupList = [];
-let groupElements = [{ groupSize: undefined, groupId: undefined, messageData: undefined }];
 let delay;
 
 async function timer(this_) {
-  for (let i = 1; i < groupElements.length; i += 1) {
-    this_.emit('data', messages.newMessageWithBody(groupElements[i]));
+  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 incomingData = {};
+    results.messages.forEach((message) => {
+      incomingData[message.messageId] = message.messageData;
+    });
+
+    // eslint-disable-next-line no-await-in-loop
+    await storage.deleteObjectById(groupList[i]);
+
+    // eslint-disable-next-line no-await-in-loop
+    await this_.emit('data', messages.newMessageWithBody({
+      groupSize: results.messages.length,
+      groupId: results.messages[0].groupId,
+      messageData: incomingData,
+    }));
   }
 }
 
@@ -72,7 +87,7 @@ async function processAction(msg, cfg) {
     Currently the group has ${messagesNumberSeen} of ${messageGroupSize} message(s).`,
   );
 
-  // when group sized is defined || when both group size and delay timer are defined
+  // when grouSized option is selected
   if (mode === 'groupSize') {
     if (messagesNumberSeen >= messageGroupSize) {
       parsedMessageGroup.messages.forEach((message) => {
@@ -90,28 +105,14 @@ async function processAction(msg, cfg) {
 
   // When delay timer option is selected
   if (mode === 'timeout') {
-    // delay timer
     delay = msg.body.timersec;
     delay = (msg.body.timersec >= 40000) ? 40000 : msg.body.timersec;
     clearTimeout(timeHandle);
     timeHandle = setTimeout(timer, delay, this);
-
-    if (groupList.includes(groupId)) {
-      parsedMessageGroup.messages.forEach((message) => {
-        incomingData[message.messageId] = message.messageData;
-      });
-      // update incoming messageData in current group
-      const currentGroup = groupElements.filter((x) => x.groupId === groupId);
-      currentGroup[0].groupSize = messagesNumberSeen;
-      currentGroup[0].messageData = incomingData;
-      groupElements = groupElements.filter((x) => x.groupId !== groupId);
-      groupElements.push(currentGroup[0]);
+    if (groupList.includes(messageGroupId)) {
+      // no action required
     } else {
-      parsedMessageGroup.messages.forEach((message) => {
-        incomingData[message.messageId] = message.messageData;
-      });
-      groupList.push(groupId);
-      groupElements.push({ groupSize: messagesNumberSeen, groupId, messageData: incomingData });
+      groupList.push(messageGroupId);
     }
   }
 
@@ -122,23 +123,12 @@ async function processAction(msg, cfg) {
     clearTimeout(timeHandle);
     timeHandle = setTimeout(timer, delay, this);
 
-    if (groupList.includes(groupId)) {
-      parsedMessageGroup.messages.forEach((message) => {
-        incomingData[message.messageId] = message.messageData;
-      });
-      // update incoming messageData in current group
-      const currentGroup = groupElements.filter((x) => x.groupId === groupId);
-      currentGroup[0].groupSize = messagesNumberSeen;
-      currentGroup[0].messageData = incomingData;
-      groupElements = groupElements.filter((x) => x.groupId !== groupId);
-      groupElements.push(currentGroup[0]);
+    if (groupList.includes(messageGroupId)) {
+      // no action required
     } else {
-      parsedMessageGroup.messages.forEach((message) => {
-        incomingData[message.messageId] = message.messageData;
-      });
-      groupList.push(groupId);
-      groupElements.push({ groupSize: messagesNumberSeen, groupId, messageData: incomingData });
+      groupList.push(messageGroupId);
     }
+
     if (messagesNumberSeen >= messageGroupSize) {
       parsedMessageGroup.messages.forEach((message) => {
         incomingData[message.messageId] = message.messageData;
@@ -151,8 +141,7 @@ async function processAction(msg, cfg) {
       }));
       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);
+      groupList = groupList.filter((def) => def !== messageGroupId);
     }
   }
 }
@@ -285,7 +274,7 @@ async function getMetaModel(cfg) {
       },
     });
   }
-  return true;
+  return false;
 }
 module.exports = {
   process: processAction,

spec/reassemble.spec.js

@@ -261,6 +261,10 @@ describe('Split on JSONata ', () => {
       .reply(200, { messages: [{ msg123: undefined }], messageIdsSeen: { msg123: 'msg123' } });
 
     const putMessageGroup1 = nock('https://ma.estr').put('/objects/group123').reply(200, {});
+    const getMessageGroup2 = nock('https://ma.estr')
+      .get('/objects/group123')
+      .reply(200, { messages: [{ groupId: 'group123' }], messageIdsSeen: { msg123: 'msg123' } });
+    const deleteMessageGroup = nock('https://ma.estr').delete('/objects/group123').reply(200, {});
 
     await reassemble.process.call(self, msg, { mode: 'timeout' });
 
@@ -272,7 +276,6 @@ describe('Split on JSONata ', () => {
       groupSize: 1,
       groupId: 'group123',
       messageData: {
-        msg123: undefined,
         undefined,
       },
     });
@@ -283,5 +286,7 @@ describe('Split on JSONata ', () => {
     expect(putMessageGroup.isDone()).to.equal(true);
     expect(getMessageGroup1.isDone()).to.equal(true);
     expect(putMessageGroup1.isDone()).to.equal(true);
+    expect(getMessageGroup2.isDone()).to.equal(true);
+    expect(deleteMessageGroup.isDone()).to.equal(true);
   });
 });