Merge pull request #6 from parse-server-modules/delete-on-sig

Delete subscription on SIGTERM / SIGINT

Author: flovilmart

lib/index.js

@@ -8,11 +8,11 @@ const namePrefix = 'parse-server';
 let emitter = new events.EventEmitter();
 
 if (!process.env.GOOGLE_APPLICATION_CREDENTIALS) {
-  throw "set GOOGLE_APPLICATION_CREDENTIALS variable (see https://googlecloudplatform.github.io/google-cloud-node/#/docs/pubsub/0.8.0/guides/authentication)";
+  throw 'set GOOGLE_APPLICATION_CREDENTIALS variable (see https://googlecloudplatform.github.io/google-cloud-node/#/docs/pubsub/0.8.0/guides/authentication)';
 }
 
 if (!process.env.GCLOUD_PROJECT) {
-  throw "set GCLOUD_PROJECT variable (see https://googlecloudplatform.github.io/google-cloud-node/#/docs/pubsub/0.8.0/guides/authentication)";
+  throw 'set GCLOUD_PROJECT variable (see https://googlecloudplatform.github.io/google-cloud-node/#/docs/pubsub/0.8.0/guides/authentication)';
 }
 
 /**
@@ -26,15 +26,6 @@ class Publisher {
 
   constructor(emitter) {
     this.emitter = emitter;
-
-    // process.on('SIGNINT',function(){
-    //   process.exit();
-    // });
-
-    // process.on('SIGTERM',function(){
-    //   process.exit();
-    // });
-
   }
 
   publish(channel, message) {
@@ -53,10 +44,7 @@ class Publisher {
 class Subscriber extends events.EventEmitter {
 
   constructor() {
-    super();    
-    process.on('SIGNINT',function(){
-      process.exit();
-    });
+    super();
   }
 
   subscribe(channel) {
@@ -87,31 +75,67 @@ class Subscriber extends events.EventEmitter {
     // Here we should get all the subscriptions under the channel topic and unsubscribe them
   }
 
-  _createSubscription(topic,channel) {
+  _createSubscription(topic, channel) {
 
     const subscriptionUUID = uuid.v1();
     var subscriptionName = `${namePrefix}-${channel}-${subscriptionUUID}`;
 
     topic.subscribe(subscriptionName, (err, subscription) => {
+
       if (err) {
-        console.log(`Failed to create subscription ${err}`);
+        console.error(`Failed to create subscription ${err}`);
         return;
       }
 
       console.log(`Subscription ${subscription.name} created.`);
-      process.on('SIGTERM', () => {
-        subscription.delete();
-      });
-      process.on('SIGINT', () => {
-        subscription.delete();
-      });
-      subscription.on('message', (message) => {
-        if (message.ackId) {
-          message.ack();
+      
+      function deleteSubscription() {
+        removeListeners();
+        console.log('Subscriber: Signal received, deleting subscription');
+        subscription.delete().then(() => {
+          console.log('Subscriber: subscription deleted...');
+        }, (err) => {
+          console.error(`Subscriber: Error deleting subscription`, err);
+        });
+      }
+      
+      function messageHandler(message) {
+          if (message.ackId) {
+            message.ack();
+          }
+          let topicName = `${namePrefix}-${channel}`;
+          this.emit('message', channel , message.data);
+      };
+
+      function errorHandler(err) {
+        console.error('Subscriber Error: ', err);
+        // Handle when subscription gets deleted
+        // Attempt to recreate it
+        if (err.code == 404) {
+          console.log('Subscriber: will recreate subscription');
+          removeListeners();
+          this._createSubscription(topic, channel);
         }
-        let topicName = `${namePrefix}-${channel}`;
-        this.emit('message', channel , message.data);
-      });
+      }
+
+      const onMessage = messageHandler.bind(this);
+      const onError = errorHandler.bind(this);
+      
+      // Remove the listenerds
+      function removeListeners() {
+        subscription.removeListener('message', onMessage);
+        subscription.removeListener('error', onError);
+        process.removeListener('SIGTERM', deleteSubscription);
+        process.removeListener('SIGINT', deleteSubscription);
+      }
+
+      // Handle termination, delete the subscription (require graceful shutdowm)
+      process.on('SIGTERM', deleteSubscription);
+      process.on('SIGINT', deleteSubscription);
+      
+      // Bind the subscription
+      subscription.on('message', onMessage);
+      subscription.on('error', onError);
     });
   }
 }
@@ -124,9 +148,7 @@ function createSubscriber() {
   return new Subscriber();
 }
 
-let GcpPubSub = {
+module.exports = {
   createPublisher,
   createSubscriber
-}
-
-module.exports = GcpPubSub;
+};

package.json

@@ -25,7 +25,7 @@
   },
   "homepage": "https://github.com/parse-server-modules/parse-server-gcloud-pubsub#readme",
   "dependencies": {
-    "@google-cloud/pubsub": "^0.5.0",
-    "uuid": "^2.0.3"
+    "@google-cloud/pubsub": "^0.9.0",
+    "uuid": "^3.0.1"
   }
 }