update

Author: EmptyInfinity

CHANGELOG.md

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

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",
@@ -13,21 +13,19 @@
         "link": "/components/csv/actions#read-csv-attachment"
       },
       "fields": {
-        "emitAll": {
+        "emitBehavior": {
           "label": "Emit Behavior",
           "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",
@@ -88,4 +86,4 @@
       "dynamicMetadata": true
     }
   }
-}
+}

lib/actions/read.js

@@ -26,10 +26,20 @@ 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 emitAll = cfg.emitAll === true || cfg.emitAll === 'true'
+  const { emitBehavior } = cfg;
+  const { body } = msg;
 
   // check if url provided in msg
   if (body.url && body.url.length > 0) {
@@ -60,9 +70,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 +93,12 @@ async function readCSV(msg, cfg) {
       } else {
         data = arrayToObj(chunk)
       }
-      if (emitAll) {
-        outputMsg.result.push(data)
-      } else {
+      if (emitBehavior === 'emitIndividually') {
         parseStream.pause()
         await that.emit('data', messages.newMessageWithBody(data))
         parseStream.resume()
+      } else {
+        result.push(data)
       }
     }
   }
@@ -111,10 +119,56 @@ async function readCSV(msg, cfg) {
     return
   }
 
-  if (emitAll) {
-    await this.emit('data', messages.newMessageWithBody(outputMsg))
+  if (emitBehavior === 'fetchAll') {
+    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 (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;
+}