feat: add rabbitmq as another way of receiving alive signals

package.json

@@ -14,6 +14,7 @@
     "author": "Almog Vagman Ciprut",
     "license": "ISC",
     "devDependencies": {
+        "@types/amqplib": "^0.5.13",
         "@types/body-parser": "^1.17.1",
         "@types/express": "^4.17.2",
         "@types/mongoose": "^5.5.34",
@@ -31,6 +32,7 @@
         "typescript": "^3.7.3"
     },
     "dependencies": {
+        "amqplib": "^0.5.5",
         "body-parser": "^1.19.0",
         "express": "^4.17.1",
         "mongoose": "^5.8.1"

src/aliveSignal/aliveSignal.subscribe.broker.ts

@@ -0,0 +1,20 @@
+import * as rabbit from '../utils/rabbit';
+import config from '../config';
+import AliveSignalManager from './aliveSignal.manager';
+import IAliveSignal from './aliveSignal.interface';
+
+export default class AliveSignalSubscribeBroker {
+    public static async subscribe() {
+        rabbit.subscribe(
+            config.rabbitMQ.exchange,
+            config.rabbitMQ.exchangeType,
+            config.rabbitMQ.queueName,
+            '#.aliveSignal',
+            async messageContent => {
+                const { hostname, serviceName, aliveDate, upTimeInSeconds } = messageContent;
+
+                await AliveSignalManager.create({ hostname, serviceName, aliveDate, upTimeInSeconds });
+            },
+        );
+    }
+}

src/config.ts

@@ -11,6 +11,15 @@ const config = {
     cors: {
         allowedOrigins: process.env.ALLOWED_ORIGINS ? process.env.ALLOWED_ORIGINS.split(',') : ['*'],
     },
+    rabbitMQ: {
+        host: process.env.RMQ_HOST || 'localhost',
+        port: +(process.env.RMQ_PORT || 5672),
+        password: process.env.RMQ_PASSWORD || 'guest',
+        username: process.env.RMQ_USERNAME || 'guest',
+        exchange: process.env.RMQ_EXCHANGE || 'drop-box-exchange',
+        exchangeType: process.env.RMQ_EXCHANGE_TYPE || 'topic',
+        queueName: process.env.RMQ_QUEUE_NAME || 'alive-signal-queue',
+    },
     aliveSignal: {
         expirationTimeInSeconds: +(process.env.ALIVE_SIGNAL_EXPIRATION_TIME || 60 * 60 * 24),
     },

src/index.ts

@@ -1,6 +1,8 @@
 import * as mongoose from 'mongoose';
+import * as rabbit from './utils/rabbit';
 import Server from './server';
 import config from './config';
+import AliveSignalSubscribeBroker from './aliveSignal/aliveSignal.subscribe.broker';
 
 initEventHandlers();
 
@@ -12,11 +14,13 @@ initEventHandlers();
 function initEventHandlers() {
     process.on('uncaughtException', error => {
         console.error('Unhandled Exception', error.stack);
+        rabbit.closeConnection();
         process.exit(1);
     });
 
     process.on('unhandledRejection', error => {
         console.error('Unhandled Rejection', error);
+        rabbit.closeConnection();
         process.exit(1);
     });
 
@@ -25,6 +29,7 @@ function initEventHandlers() {
             console.log('User Termination');
 
             await mongoose.disconnect();
+            rabbit.closeConnection();
             process.exit(0);
         } catch (error) {
             console.error('Failed to close MongoDB connection before server shutdown', error);
@@ -60,5 +65,8 @@ async function setMongoConnection() {
         console.log('[MongoDB] reconnected');
     });
 
+    await rabbit.connect();
+    await AliveSignalSubscribeBroker.subscribe();
+
     return mongoose.connect(config.db.connectionString, { useNewUrlParser: true });
 }