Issue 180 component updates (#23)

* add read action; test view

* add read action; test view; component.json hotfix

* add read action; test view; component.json fix action name

* add read action; test view; component.json fix read.js

* add read action; test view;

* add read action; test view;

* add read action; test view; add jsonataview for url

* add read action; test view; add username & password fields

* add read action; add placeholders processing

* add read action; add exports

* add read action; change placeholders processing

* add read action; add logs to placeholders processing

* read action; mapping body to url placeholders

* read action; mapping body to url placeholders

* read action; mapping body to url placeholders

* Going to eslint airbnb-base codestyle.

* Fixinig unit test:
- Changing input test data format from json to csv (as expected)
- Setting correct output expected number.

* add read action

* docs draft

* add response as attachment for read action

* replace extra dependencies with utils functions

* add header and required delimiter to output csv for read action

* sailor upgraded to 2.2.3
migrated to v2 circleci
added await for emit data

* Issue 178 fixing oom (#25)

* Fixing OOM - piping csv processing stream.
* Adding limitations to docs

* Update README.md

* Update README.md (#27)

Author: ShkarupaNick

README.md

@@ -13,7 +13,7 @@ attachment. It can also write a CSV file from the incoming events.
 
 ## Environment variables
 
-Component is not using any environment variables.
+Component does not has any required environment variables, but we suggest to use `EIO_REQUIRED_RAM_MB` - recommended value of allocated memory is `512` MB
 
 
 ## Credentials
@@ -29,6 +29,11 @@ This trigger will fetch the CSV file from a given URL. The address must be acces
 to the component. The fetched CSV file will be placed in the attachment part of the
 outgoing message.
 
+#### Limitations
+
+You may get `Component run out of memory and terminated.` error during run-time, that means that component needs more memory, please add
+ `EIO_REQUIRED_RAM_MB` Environment variables for csv-component in this case.
+
 ## Actions
 
 ### Read CSV attachment
@@ -61,6 +66,7 @@ for that cell. All other properties will be ignored. For example, headers
 ```
 
 will produce the following `.csv` file:
+
 ```
 foo,bar
 myfoo,mybar

circle.yml

@@ -1,13 +1,24 @@
-machine:
-  services:
-    - docker
-
-dependencies:
-  override:
-    - docker pull elasticio/appbuilder
-
-test:
-  override:
-    - chmod 700 .circleci/build_slug.sh
-    - .circleci/build_slug.sh
-
+version: 2
+jobs:
+  test:
+    docker:
+      - image: circleci/node:8-stretch
+    steps:
+      - checkout
+      - restore_cache:
+          key: dependency-cache-{{ checksum "package.json" }}
+      - run:
+          name: Installing Dependencies
+          command: npm install
+      - save_cache:
+          key: dependency-cache-{{ checksum "package.json" }}
+          paths:
+            - node_modules
+      - run:
+          name: Running Mocha Tests
+          command: npm test
+workflows:
+  version: 2
+  build_and_test:
+    jobs:
+      - test

lib/actions/write.js

@@ -43,7 +43,7 @@ exports.init = function init(cfg) {
     });
 };
 
-exports.process = function ProcessAction(msg, cfg) {
+exports.process = async function ProcessAction(msg, cfg) {
     // eslint-disable-next-line consistent-this
     const self = this;
 
@@ -70,7 +70,7 @@ exports.process = function ProcessAction(msg, cfg) {
             rowCount = 0;
 
             console.log('Emitting message %j', messageToEmit);
-            self.emit('data', messageToEmit);
+            yield self.emit('data', messageToEmit);
 
             yield exports.init(cfg);
         });
@@ -85,5 +85,5 @@ exports.process = function ProcessAction(msg, cfg) {
     console.log(`Writing Row: ${JSON.stringify(row)}`);
     stringifier.write(row);
     rowCount++;
-    this.emit('end');
+    await this.emit('end');
 };

lib/read.js

@@ -6,6 +6,7 @@ const messages = require('elasticio-node').messages;
 const moment = require('moment');
 const debug = require('debug')('csv');
 const request = require('request');
+const { Writable } = require('stream');
 
 const formatters = {};
 
@@ -56,7 +57,7 @@ function createRowMessage(row, columns) {
     return messages.newMessageWithBody(body);
 }
 
-function ProcessRead(msg, cfg) {
+async function ProcessRead(msg, cfg) {
     let csvURL = cfg.url;
     const that = this;
     let index = 0;
@@ -73,49 +74,61 @@ function ProcessRead(msg, cfg) {
         } else {
             console.error('URL of the CSV is missing');
             that.emit('error', 'URL of the CSV is missing');
-            return that.emit('end');
+            return await that.emit('end');
         }
     }
     const parser = csv.parse({
         delimiter: separator
     });
 
-    parser.on('readable', function onReadable() {
-        let record;
-        // eslint-disable-next-line no-cond-assign
-        while (record = parser.read()) {
-            debug('Have got a row #%s', index);
+    let ended = false;
+
+    class CsvWriter extends Writable {
+        async write(chunk, encoding, callback) {
+            parser.pause();
+            debug('Processing %d row...', index);
+            debug('Memory usage: %d Mb', process.memoryUsage().heapUsed / 1024 / 1024);
+
             if (index >= startRow) {
-                const msg = createRowMessage(record, cfg.reader.columns);
-                that.emit('data', msg);
+                const msg = createRowMessage(chunk, cfg.reader.columns);
+                await that.emit('data', msg);
             } else {
                 debug('Row #%s is skipped based on configuration', index);
             }
             index++;
+            parser.resume();
         }
-    });
 
-    parser.on('finish', function fireEnd() {
-        debug('Number of lines: ' + index);
-        that.emit('end');
-    });
+        async end(chunk, encoding, callback) {
+            debug('Processing csv writer end event...');
+            debug('Memory usage: %d Mb', process.memoryUsage().heapUsed / 1024 / 1024);
 
-    parser.on('error', function emitError(error) {
-        debug('Error reported by CSV read', error);
-        that.emit('error', error);
-        that.emit('end');
-    });
+            debug('Number of lines: ' + index);
+            await that.emit('end');
+            ended = true;
+        }
+    }
+
+    const writer = new CsvWriter();
 
     debug('Sending GET request to url=%s', csvURL);
     request.get(csvURL)
-        .on('response', function onResponse(response) {
+        .on('response', async function onResponse(response) {
             debug('Have got response status=%s headers=%j', response.statusCode, response.headers);
             if (response.statusCode !== 200) {
-                that.emit('error', 'Unexpected response code code=' + response.statusCode);
+                await that.emit('error', 'Unexpected response code code=' + response.statusCode);
+                ended = true;
                 throw Error('Unexpected response code code=' + response.statusCode);
             }
         })
-        .pipe(parser);
+        .pipe(parser).pipe(writer);
+
+    // Need to wait for processing all messages
+    while (!ended) {
+        await new Promise((resolve, reject) => {
+            setTimeout(() => { resolve(); }, 0);
+        });
+    }
 }
 
 exports.process = ProcessRead;