Merge pull request #3 from elasticio/add-emit-behaviour

## 3.1.0 (March 3, 2022)
* Added `Emit Batch` behavior for `Read CSV attachment` action

Author: stas-fomenko

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.1.0 (March 3, 2022)
+* Added `Emit Batch` behavior for `Read CSV attachment` action
+
 ## 3.0.0 (July 9, 2021)
 * Deleted trigger:  
   - `Read CSV attachment`

README.md

@@ -48,14 +48,15 @@ To configure this action the following fields can be used:
 
 #### Config Fields
 
-*   `Emit Behavior` - this selector configures output behavior of the component. If the option is `Fetch All` - the component emits an array of messages, otherwise (`Emit Individually`) - the component emits a message per row.
+*   `Emit Behavior` - this selector configures output behavior of the component. If the option is `Fetch All` - the component emits an array of messages; `Emit Individually` - the component emits a message per row; `Emit Batch` - component will produce a series of message where each message has an array of max length equal to the `Batch Size`;
 
 #### Input Metadata
 
 *   `URL` - We will fetch this URL and parse it as CSV file
 *   `Contains headers` - if true, the first row of parsed data will be interpreted as field names, false by default.
 *   `Delimiter` - The delimiting character. Leave blank to auto-detect from a list of most common delimiters.
 *   `Convert Data types` - numeric, date and boolean data will be converted to their type instead of remaining strings, false by default.
+If `Emit Behavior` equals to `Emit Batch` - new field appears: `Batch Size` - max length of array for each message
 
 ### Create CSV From Message Stream
 

component.json

@@ -3,7 +3,7 @@
   "description": "A comma-separated values (CSV) file stores tabular data (numbers and text) in plain-text form",
   "docsUrl": "https://github.com/elasticio/csv-component",
   "buildType": "docker",
-  "version": "3.0.0",
+  "version": "3.1.0",
   "actions": {
     "read_action": {
       "main": "./lib/actions/read.js",
@@ -18,16 +18,14 @@
           "required": true,
           "viewClass": "SelectView",
           "model": {
-            "true": "Fetch All",
-            "false": "Emit Individually"
+            "fetchAll": "Fetch All",
+            "emitIndividually": "Emit Individually",
+            "emitBatch": "Emit Batch"
           },
           "prompt": "Select Emit Behavior"
         }
       },
-      "metadata": {
-        "in": "./lib/schemas/read.in.json",
-        "out": {}
-      }
+      "dynamicMetadata": true
     },
     "write_from_stream": {
       "main": "./lib/actions/writeStream.js",

lib/actions/read.js

@@ -26,10 +26,19 @@ async function errHelper(text) {
   await this.emit('end')
 }
 
+function sliceIntoChunks(arr, chunkSize) {
+  const res = [];
+  for (let i = 0; i < arr.length; i += chunkSize) {
+    const chunk = arr.slice(i, i + chunkSize);
+    res.push(chunk);
+  }
+  return res;
+}
+
 async function readCSV(msg, cfg) {
   const that = this
-  const emitAll = cfg.emitAll === true || cfg.emitAll === 'true'
-  const { body } = msg
+  const emitBehavior = cfg.emitAll;
+  const { body } = msg;
 
   // check if url provided in msg
   if (body.url && body.url.length > 0) {
@@ -60,9 +69,7 @@ async function readCSV(msg, cfg) {
   }
 
   // if set "Fetch All" create object with results
-  const outputMsg = {
-    result: [],
-  }
+  const result = [];
 
   let dataStream
   const parseStream = papa.parse(papa.NODE_STREAM_INPUT, parseOptions)
@@ -85,12 +92,12 @@ async function readCSV(msg, cfg) {
       } else {
         data = arrayToObj(chunk)
       }
-      if (emitAll) {
-        outputMsg.result.push(data)
-      } else {
+      if (emitBehavior === 'emitIndividually' || cfg.emitAll === false || cfg.emitAll === 'false') {
         parseStream.pause()
         await that.emit('data', messages.newMessageWithBody(data))
         parseStream.resume()
+      } else {
+        result.push(data)
       }
     }
   }
@@ -111,10 +118,56 @@ async function readCSV(msg, cfg) {
     return
   }
 
-  if (emitAll) {
-    await this.emit('data', messages.newMessageWithBody(outputMsg))
+  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);
+    // eslint-disable-next-line no-plusplus
+    for (let i = 0; i < chunks.length; i++) {
+      // eslint-disable-next-line no-await-in-loop
+      await this.emit('data', messages.newMessageWithBody({ result: chunks[i] }))
+    }
   }
   this.logger.info(`Complete, memory used: ${process.memoryUsage().heapUsed / 1024 / 1024} Mb`)
 }
 
-module.exports.process = readCSV
+module.exports.process = readCSV;
+module.exports.getMetaModel = async function getMetaModel(cfg) {
+  const meta = {
+    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: {}
+  };
+
+  if (cfg.emitBehavior === 'emitBatch') {
+    meta.in.properties.batchSize = {
+      title: 'Batch Size',
+      type: 'number',
+      required: true
+    }
+  }
+  return meta;
+}