Fix memory leak on Emit Batch (#12)

Fix memory leak on `Emit Batch`

Author: if0s

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.1.4 (May 06, 2022)
+* Fix memory leak on `Emit Batch` behavior for `Read CSV attachment` action
+
 ## 3.1.3 (April 22, 2022)
 * Fix a bug when emit strategy 'Emit Batch' did not process correctly
 

README.md

@@ -58,6 +58,15 @@ To configure this action the following fields can be used:
 *   `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
 
+#### Output Metadata
+- For `Fetch page` and `Emit Batch`: An object with key ***result*** that has an array as its value
+- For `Emit Individually`:  Each object fill the entire message
+
+#### Limitations
+* If you use `Fetch All` then component needs to store whole file and object in memory that cause big memory usage
+* In `Emit Batch` use wisely `Batch Size`, bigger number cause bigger memory usage
+* Possible exception: `[ERR_STREAM_PREMATURE_CLOSE]` could be thrown when flow stopped before finish emiting all data in file, as stream stopped
+
 ### Create CSV From Message Stream
 
 This action will combine multiple incoming events into a CSV file until there is a gap

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.3",
+  "version": "3.1.4",
   "actions": {
     "read_action": {
       "main": "./lib/actions/read.js",

lib/actions/read.js

@@ -26,20 +26,18 @@ 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 emitBehavior = cfg.emitAll;
   const { body } = msg;
 
+  let batchSize;
+  if (emitBehavior === 'emitBatch') {
+    batchSize = body.batchSize;
+    if (!isPositiveInteger(batchSize)) {
+      throw new Error("'batchSize' must be a positive integer!");
+    }
+  }
   // check if url provided in msg
   if (body.url && body.url.length > 0) {
     this.logger.info('URL found')
@@ -83,7 +81,8 @@ async function readCSV(msg, cfg) {
     this.emit('end')
     return
   }
-  // control of node data stream
+
+  const buf = [];
   class CsvWriter extends Writable {
     async write(chunk) {
       let data = {}
@@ -92,9 +91,14 @@ async function readCSV(msg, cfg) {
       } else {
         data = arrayToObj(chunk)
       }
-      if (emitBehavior === 'emitIndividually' || cfg.emitAll === false || cfg.emitAll === 'false') {
+      if (emitBehavior === 'emitIndividually' || cfg.emitAll === false || cfg.emitAll === 'false' || emitBehavior === 'emitBatch') {
         parseStream.pause()
-        await that.emit('data', messages.newMessageWithBody(data))
+        if (emitBehavior === 'emitBatch') {
+          buf.push(data);
+          if (buf.length >= batchSize) await that.emit('data', messages.newMessageWithBody({ result: buf.splice(0, batchSize) }))
+        } else {
+          await that.emit('data', messages.newMessageWithBody(data))
+        }
         parseStream.resume()
       } else {
         result.push(data)
@@ -120,17 +124,8 @@ 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 { 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
-      await this.emit('data', messages.newMessageWithBody({ result: chunks[i] }))
-    }
+  } else if (emitBehavior === 'emitBatch' && buf.length > 0) {
+    await that.emit('data', messages.newMessageWithBody({ result: buf }))
   }
   this.logger.info(`Complete, memory used: ${process.memoryUsage().heapUsed / 1024 / 1024} Mb`)
 }