Adding waiting for timer mechanism

Author: denyshld

lib/actions/write.js

@@ -19,6 +19,8 @@ let rowCount = 0;
 let ax;
 let putUrl;
 
+let readyFlag = false;
+
 exports.init = async function init(cfg) {
   const delimiter = cfg.writer.separator || ',';
   const header = cfg.includeHeaders !== 'No';
@@ -45,17 +47,25 @@ exports.init = async function init(cfg) {
   console.log('CSV file to be uploaded file to uri=%s', putUrl);
   ax = axios.create();
   util.addRetryCountInterceptorToAxios(ax);
+  readyFlag = true;
 };
 
 exports.process = async function ProcessAction(msg, cfg) {
   // eslint-disable-next-line consistent-this
   const self = this;
 
+  while (!readyFlag) {
+    // eslint-disable-next-line no-loop-func,no-await-in-loop
+    await new Promise(resolve => timeout(resolve, 100));
+  }
+
   if (timeout) {
     clearTimeout(timeout);
   }
 
   timeout = setTimeout(async () => {
+    readyFlag = false;
+
     console.log('Closing the stream due to inactivity');
 
     const finalRowCount = rowCount;
@@ -68,6 +78,7 @@ exports.process = async function ProcessAction(msg, cfg) {
       maxContentLength: REQUEST_MAX_CONTENT_LENGTH,
     });
     stringifier.end();
+
     const messageToEmit = messages.newMessageWithBody({
       rowCount: finalRowCount,
     });
@@ -80,7 +91,6 @@ exports.process = async function ProcessAction(msg, cfg) {
     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;