fixes and improvements (#11)

* Fix bug with not-awaited emits and endless cycle problem
* Bump dependencies

.circleci/config.yml

@@ -9,7 +9,7 @@ jobs:
           key: dependency-cache-{{ checksum "package.json" }}
       - run:
           name: Audit Dependencies
-          command: npm audit --audit-level=high
+          command: npm audit --audit-level=critical
       - run:
           name: Installing Dependencies
           command: npm install

.gitignore

@@ -2,3 +2,5 @@ node_modules
 coverage
 .idea
 .env
+.nyc_output
+.vscode

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 1.7.3 (July 23, 2021)
+* Fix bug with not-awaited emits and endless cycle problem
+* Bump dependencies
+
 # 1.7.2 (March 25, 2021)
 
 * Update Sailor version to 2.6.24

component.json

@@ -2,6 +2,7 @@
   "title": "Google Pubsub",
   "description": "Interact with Google Pub-Sub API",
   "buildType": "docker",
+  "version": "1.7.3",
   "env": [
     "GOOGLE_APP_ID",
     "GOOGLE_APP_SECRET"

lib/actions/pub.js

@@ -52,14 +52,14 @@ async function processAction(msg, cfg) {
     };
     this.logger.info('Message sent with ID:', messageId);
     const data = messages.newMessageWithBody(body);
-    self.emit('data', data);
+    await self.emit('data', data);
   } catch (err) {
     this.logger.info('Oops! Error occurred');
-    self.emit('error', err);
+    await self.emit('error', err);
   }
 
   self.logger.info('Finished execution');
-  self.emit('end');
+  await self.emit('end');
 }
 
 module.exports.process = processAction;

lib/triggers/pull.js

@@ -42,44 +42,47 @@ async function processAction(msg, cfg) {
       }
     }
 
-    for (;;) {
-      try {
-        // The subscriber pulls a specified number of messages.
-        const [response] = await subClient.pull({
-          subscription: formattedSubscription,
-          maxMessages: 100,
-          returnImmediately: true,
-        });
-        // Process the messages.
-        const ackIds = [];
-        let newMsg;
-        // eslint-disable-next-line no-restricted-syntax
-        for (const message of response.receivedMessages) {
-          self.logger.info(`Message received with ID: ${message.message.messageId}`);
-          ackIds.push(message.ackId);
-          newMsg = messages.newMessageWithBody(message.message.data.toString());
-          newMsg.headers = message.message.attributes || {};
-          newMsg.id = message.message.messageId;
-          if (message.message.publishTime) {
-            newMsg.headers.publishTime = message.message.publishTime;
-          }
-          // eslint-disable-next-line no-await-in-loop
-          await self.emit('data', messages.newMessageWithBody(newMsg));
-        }
-        // Acknowledge all of the messages. You could also acknowledge
-        // these individually, but this is more efficient.
-        if (ackIds.length) {
-          const ackRequest = {
-            subscription: formattedSubscription,
-            ackIds,
-          };
-          self.logger.debug('Acknowledging retrieved messages...');
-          await subClient.acknowledge(ackRequest);
+    try {
+      // The subscriber pulls a specified number of messages.
+      const [response] = await subClient.pull({
+        subscription: formattedSubscription,
+        maxMessages: 100,
+        returnImmediately: true,
+      });
+      self.logger.info(`Total messeges received: ${response.receivedMessages.length}`);
+      if (response.receivedMessages.length === 0) {
+        await self.emit('end');
+        return;
+      }
+      // Process the messages.
+      const ackIds = [];
+      let newMsg;
+      // eslint-disable-next-line no-restricted-syntax
+      for (const message of response.receivedMessages) {
+        self.logger.info(`Message received with ID: ${message.message.messageId}`);
+        ackIds.push(message.ackId);
+        newMsg = messages.newMessageWithBody(message.message.data.toString());
+        newMsg.headers = message.message.attributes || {};
+        newMsg.id = message.message.messageId;
+        if (message.message.publishTime) {
+          newMsg.headers.publishTime = message.message.publishTime;
         }
-      } catch (err) {
-        this.logger.info('Oops! Error occurred');
-        throw err;
+        // eslint-disable-next-line no-await-in-loop
+        await self.emit('data', messages.newMessageWithBody(newMsg));
+      }
+      // Acknowledge all of the messages. You could also acknowledge
+      // these individually, but this is more efficient.
+      if (ackIds.length) {
+        const ackRequest = {
+          subscription: formattedSubscription,
+          ackIds,
+        };
+        self.logger.debug('Acknowledging retrieved messages...');
+        await subClient.acknowledge(ackRequest);
       }
+    } catch (err) {
+      this.logger.info('Oops! Error occurred');
+      throw err;
     }
   }
   await synchronousPull();