Fix emit batch strategy (#9)

elasticio · Apr 22, 2022 · a19e8ba · a19e8ba
1 parent 7a0436c
commit a19e8ba
Show file tree

Hide file tree

Showing 5 changed files with 133 additions and 4 deletions.
diff --git a/.eslintrc.js b/.eslintrc.js
@@ -5,5 +5,6 @@ module.exports = {
   },
   rules: {
     'import/extensions': 'off',
+    "no-use-before-define": 0,
   },
 };
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,6 @@
+## 3.1.3 (April 22, 2022)
+* Fix a bug when emit strategy 'Emit Batch' did not process correctly
+
 ## 3.1.2 (April 14, 2022)
 * Update `component-commons-library` to read and upload attachments through `Maester`
 * Update `elasticio-sailor-nodejs` to v2.6.27

diff --git a/component.json b/component.json
@@ -2,7 +2,7 @@
   "title": "CSV",
   "description": "A comma-separated values (CSV) file stores tabular data (numbers and text) in plain-text form",
   "docsUrl": "https://github.com/elasticio/csv-component",
-  "version": "3.1.2",
+  "version": "3.1.3",
   "actions": {
     "read_action": {
       "main": "./lib/actions/read.js",
@@ -85,4 +85,4 @@
       "dynamicMetadata": true
     }
   }
-}
+}
diff --git a/lib/actions/read.js b/lib/actions/read.js
@@ -121,7 +121,11 @@ async function readCSV(msg, cfg) {
   if (emitBehavior === 'fetchAll' || cfg.emitAll === true || cfg.emitAll === 'true') {
     await this.emit('data', messages.newMessageWithBody({ result }))
   } else if (emitBehavior === 'emitBatch') {
-    const chunks = sliceIntoChunks(result, body.batchSize);
+    const { batchSize } = body;
+    if (!isPositiveInteger(batchSize)) {
+      throw new Error("'batchSize' must be a positive integer!");
+    }
+    const chunks = sliceIntoChunks(result, batchSize);
     // eslint-disable-next-line no-plusplus
     for (let i = 0; i < chunks.length; i++) {
       // eslint-disable-next-line no-await-in-loop
@@ -162,7 +166,7 @@ module.exports.getMetaModel = async function getMetaModel(cfg) {
     out: {}
   };
 
-  if (cfg.emitBehavior === 'emitBatch') {
+  if (cfg.emitAll === 'emitBatch') {
     meta.in.properties.batchSize = {
       title: 'Batch Size',
       type: 'number',
@@ -171,3 +175,7 @@ module.exports.getMetaModel = async function getMetaModel(cfg) {
   }
   return meta;
 }
+
+function isPositiveInteger(input) {
+  return Number.isSafeInteger(input) && input > 0;
+}
diff --git a/spec/read.spec.js b/spec/read.spec.js
@@ -62,6 +62,87 @@ describe('CSV Read component', async () => {
     });
   });
 
+  describe('emitAll: emitBatch', async () => {
+    it('should contain batchSize field', async () => {
+      cfg = {
+        emitAll: 'emitBatch',
+      };
+      const expectedMetadata = {
+        in: {
+          type: 'object',
+          properties: {
+            url: {
+              type: 'string',
+              required: true,
+              title: 'URL',
+            },
+            header: {
+              type: 'boolean',
+              required: false,
+              title: 'Contains headers',
+            },
+            delimiter: {
+              type: 'string',
+              required: false,
+              title: 'Delimiter',
+            },
+            dynamicTyping: {
+              type: 'boolean',
+              required: false,
+              title: 'Convert Data types',
+            },
+            batchSize: {
+              title: 'Batch Size',
+              type: 'number',
+              required: true,
+            },
+          },
+        },
+        out: {},
+      };
+      const metadata = await readCSV.getMetaModel(cfg);
+      expect(metadata).to.deep.equal(expectedMetadata);
+    });
+  });
+
+  describe('emitAll: fetchAll', async () => {
+    it('should not contain batchSize field', async () => {
+      cfg = {
+        emitAll: 'fetchAll',
+      };
+      const expectedMetadata = {
+        in: {
+          type: 'object',
+          properties: {
+            url: {
+              type: 'string',
+              required: true,
+              title: 'URL',
+            },
+            header: {
+              type: 'boolean',
+              required: false,
+              title: 'Contains headers',
+            },
+            delimiter: {
+              type: 'string',
+              required: false,
+              title: 'Delimiter',
+            },
+            dynamicTyping: {
+              type: 'boolean',
+              required: false,
+              title: 'Convert Data types',
+            },
+          },
+        },
+        out: {},
+      };
+      const metadata = await readCSV.getMetaModel(cfg);
+      expect(metadata).to.deep.equal(expectedMetadata);
+    });
+  });
+
   it('One file', async () => {
     msg.body = {
       url: 'http://test.env.mock/formats.csv',
@@ -133,6 +214,42 @@ describe('CSV Read component', async () => {
       .to.equal(2.71828); // Number
   });
 
+  it('emitBatch: true, batchSize is negative', async () => {
+    msg.body = {
+      url: 'http://test.env.mock/formats.csv',
+      header: true,
+      dynamicTyping: true,
+      batchSize: -5,
+    };
+    cfg = {
+      emitAll: 'emitBatch',
+    };
+    context.emit = sinon.spy();
+    try {
+      await readCSV.process.call(context, msg, cfg);
+    } catch (err) {
+      expect(err.message).to.be.equal('\'batchSize\' must be a positive integer!');
+    }
+  });
+
+  it('emitBatch: true, batchSize is string', async () => {
+    msg.body = {
+      url: 'http://test.env.mock/formats.csv',
+      header: true,
+      dynamicTyping: true,
+      batchSize: 'asd',
+    };
+    cfg = {
+      emitAll: 'emitBatch',
+    };
+    context.emit = sinon.spy();
+    try {
+      await readCSV.process.call(context, msg, cfg);
+    } catch (err) {
+      expect(err.message).to.be.equal('\'batchSize\' must be a positive integer!');
+    }
+  });
+
   it('emitAll: emitIndividually, header: false, dynamicTyping: false', async () => {
     msg.body = {
       url: 'http://test.env.mock/formats.csv',