Improve writing logic. Add unit test basement

Author: denyshld

.gitignore

@@ -27,4 +27,6 @@ build/Release
 
 # Dependency directory
 # https://www.npmjs.org/doc/misc/npm-faq.html#should-i-check-my-node_modules-folder-into-git
-node_modules
+node_modules
+
+.env

lib/actions/write.js

@@ -19,8 +19,6 @@ let rowCount = 0;
 let ax;
 let putUrl;
 
-const resolveList = [];
-
 exports.init = async function init(cfg) {
   const delimiter = cfg.writer.separator || ',';
   const header = cfg.includeHeaders !== 'No';
@@ -41,7 +39,7 @@ exports.init = async function init(cfg) {
   signedUrl = await client.resources.storage.createSignedUrl();
   // signedUrl = {
   //   put_url: 'https://examlple.mock/putUrl',
-  //   get_url: 'https://examlple.mock/putUrl',
+  //   get_url: 'https://examlple.mock/getUrl',
   // };
   putUrl = signedUrl.put_url;
   console.log('CSV file to be uploaded file to uri=%s', putUrl);
@@ -55,53 +53,45 @@ exports.process = async function ProcessAction(msg, cfg) {
 
   if (timeout) {
     clearTimeout(timeout);
-    resolveList.forEach((res) => {
-      res();
-    });
   }
 
-  const result = new Promise((resolve) => {
-    resolveList.push(resolve);
+  timeout = setTimeout(async () => {
+    console.log('Closing the stream due to inactivity');
 
-    timeout = setTimeout(async () => {
-      console.log('Closing the stream due to inactivity');
-
-      const finalRowCount = rowCount;
-      console.log('The resulting CSV file contains %s rows', finalRowCount);
-      ax.put(putUrl, stringifier, {
-        method: 'PUT',
-        timeout: REQUEST_TIMEOUT,
-        retry: REQUEST_MAX_RETRY,
-        delay: REQUEST_RETRY_DELAY,
-        maxContentLength: REQUEST_MAX_CONTENT_LENGTH,
-      });
-      stringifier.end();
-      const messageToEmit = messages.newMessageWithBody({
-        rowCount: finalRowCount,
-      });
-      const fileName = `${messageToEmit.id}.csv`;
-      messageToEmit.attachments[fileName] = {
-        'content-type': 'text/csv',
-        url: signedUrl.get_url,
-      };
-      signedUrl = null;
-      rowCount = 0;
-      console.log('Emitting message %j', messageToEmit);
-      await self.emit('data', messageToEmit);
-      await exports.init(cfg);
-      resolve();
-    }, TIMEOUT_BETWEEN_EVENTS);
+    const finalRowCount = rowCount;
+    console.log('The resulting CSV file contains %s rows', finalRowCount);
+    ax.put(putUrl, stringifier, {
+      method: 'PUT',
+      timeout: REQUEST_TIMEOUT,
+      retry: REQUEST_MAX_RETRY,
+      delay: REQUEST_RETRY_DELAY,
+      maxContentLength: REQUEST_MAX_CONTENT_LENGTH,
+    });
+    stringifier.end();
+    const messageToEmit = messages.newMessageWithBody({
+      rowCount: finalRowCount,
+    });
+    const fileName = `${messageToEmit.id}.csv`;
+    messageToEmit.attachments[fileName] = {
+      'content-type': 'text/csv',
+      url: signedUrl.get_url,
+    };
+    signedUrl = null;
+    rowCount = 0;
+    console.log('Emitting message %j', messageToEmit);
+    await self.emit('data', messageToEmit);
+    await exports.init(cfg);
+  }, TIMEOUT_BETWEEN_EVENTS);
 
-    let row = msg.body.writer;
-    console.log(`Incoming data: ${JSON.stringify(row)}`);
-    if (cfg.writer.columns) {
-      const columns = Object.keys(_.keyBy(cfg.writer.columns, 'property'));
-      row = _.pick(row, columns);
-    }
-    console.log(`Writing Row: ${JSON.stringify(row)}`);
-    stringifier.write(row);
-    rowCount += 1;
-  });
+  let row = msg.body.writer;
+  console.log(`Incoming data: ${JSON.stringify(row)}`);
+  if (cfg.writer.columns) {
+    const columns = Object.keys(_.keyBy(cfg.writer.columns, 'property'));
+    row = _.pick(row, columns);
+  }
+  console.log(`Writing Row: ${JSON.stringify(row)}`);
+  stringifier.write(row);
+  rowCount += 1;
 
-  await result;
+  await self.emit('end');
 };

package-lock.json

@@ -46,6 +46,7 @@
   "devDependencies": {
     "chai": "4.1.1",
     "chai-as-promised": "7.1.1",
+    "dotenv": "5.0.0",
     "eslint": "5.16.0",
     "eslint-config-airbnb-base": "^13.1.0",
     "eslint-plugin-import": "^2.17.3",

package.json

@@ -0,0 +1,83 @@
+/* eslint-disable no-unused-vars */
+const chai = require('chai');
+const chaiAsPromised = require('chai-as-promised');
+const fs = require('fs');
+const nock = require('nock');
+const sinon = require('sinon');
+
+chai.use(chaiAsPromised);
+const { expect } = require('chai');
+
+const write = require('../lib/actions/write.js');
+
+describe('CSV Write component', function () {
+  this.timeout(15000);
+
+  let emitter;
+  let cfg;
+
+  beforeEach(() => {
+    emitter = {
+      emit: sinon.spy(),
+    };
+  });
+
+  before(() => {
+    if (fs.existsSync('.env')) {
+      // eslint-disable-next-line global-require
+      require('dotenv').config();
+    }
+
+    cfg = {
+      writer: {
+        columns: [
+          { property: 'header1' },
+          { property: 'header2' },
+        ],
+      },
+    };
+
+    nock('http://api-service.platform.svc.cluster.local.:9000/', { encodedQueryParams: true })
+      .post('/v2/resources/storage/signed-url')
+      .reply(200, { put_url: 'https://examlple.mock/putUrl', get_url: 'https://examlple.mock/getUrl' });
+
+    nock('https://examlple.mock')
+      .put('/putUrl')
+      .reply(200, {});
+  });
+
+  it('should parse simple string rows', async (done) => {
+    // await write.init(cfg);
+
+    const msg1 = {
+      body: {
+        writer: {
+          columns: [
+            { property: 'text11' },
+            { property: 'text12' },
+          ],
+        },
+      },
+    };
+    // await write.process.call(emitter, msg1, cfg);
+
+    const msg2 = {
+      body: {
+        reader: {
+          columns: [
+            { property: 'text21' },
+            { property: 'text22' },
+          ],
+        },
+      },
+    };
+    // await write.process.call(emitter, msg2, cfg);
+
+    // expect(emitter.emit.getCalls().length).to.equal(2);
+
+    // await new Promise(resolve => setTimeout(resolve, 12000));
+    // expect(emitter.emit.getCalls().length).to.equal(3);
+
+    done();
+  });
+});