Merge pull request #11 from banananyo/emit-event-when-connection-closed

feat: emit event on connection closed

Author: c18s

lib/constants/amqp.constants.ts

@@ -1,3 +1,4 @@
 export const AMQP_MODULE_OPTIONS = 'AMQP_MODULE_OPTIONS';
 export const AMQP_CONSUMER_METADATA = 'AMQP_CONSUMER_METADATA';
 export const AMQP_CONNECTION_RECONNECT = 'AMQP_CONNECTION_RECONNECT';
+export const AMQP_CONNECTION_DISCONNECTED = 'AMQP_CONNECTION_DISCONNECTED';

lib/services/amqp.service.ts

@@ -4,7 +4,7 @@ import { EventEmitter } from 'events';
 import { hostname } from 'os';
 import { AwaitableSender, Connection, ConnectionEvents, Container, EventContext, Receiver, ReceiverEvents, SenderEvents } from 'rhea-promise';
 
-import { AMQP_CONNECTION_RECONNECT } from '../constants';
+import { AMQP_CONNECTION_DISCONNECTED, AMQP_CONNECTION_RECONNECT } from '../constants';
 import { AMQPModuleOptions, CreateReceiverOptions, CreateSenderOptions } from '../interfaces';
 import { ErrorMessage, getConnectionToken, getLogger, parseURL } from '../utils';
 
@@ -46,6 +46,10 @@ export class AMQPService {
     connection.on(ConnectionEvents.disconnected, (context: EventContext) => {
       const error = [`Connection closed by peer: ${connectionToken}`, ErrorMessage.fromContext(context)];
       this.logger.warn(...error.filter(e => e));
+      const emitted = AMQPService.eventEmitter.emit(AMQP_CONNECTION_DISCONNECTED);
+      if (!emitted) {
+        this.logger.warn('disconnect event not emitted');
+      }
     });
     connection.on(ConnectionEvents.connectionClose, (context: EventContext) => {
       const error = `Connection closed: ${connectionToken}`;
@@ -54,6 +58,22 @@ export class AMQPService {
       } else {
         this.logger.warn(error);
       }
+      const timeoutHandler = setTimeout(async () => {
+        (context.connection as any)._connection.dispatch(ConnectionEvents.disconnected, void 0);
+        await context.connection
+          .open()
+          .then(() => {
+            this.logger.silly('connection successfully reopened');
+            const emitted = AMQPService.eventEmitter.emit(AMQP_CONNECTION_RECONNECT);
+            if (!emitted) {
+              this.logger.warn('reconnect event not emitted');
+            }
+          })
+          .catch(error => {
+            this.logger.error(`reopening connection failed with error: ${error.message}`, error);
+          });
+        clearTimeout(timeoutHandler);
+      }, 1000);
     });
     try {
       await connection.open();

src/app.service.ts

@@ -1,5 +1,5 @@
 import { Logger } from '@dollarsign/logger';
-import { Consumer, MessageControl, ProducerService, SendOptions } from '@dollarsign/nestjs-amqp';
+import { AMQP_CONNECTION_DISCONNECTED, AMQPService, Consumer, MessageControl, ProducerService, SendOptions } from '@dollarsign/nestjs-amqp';
 import { delay } from '@dollarsign/utils';
 import { Injectable } from '@nestjs/common';
 
@@ -13,7 +13,15 @@ export class AppService {
     displayFilePath: false,
   });
 
-  constructor(private readonly producer: ProducerService) {}
+  constructor(private readonly producer: ProducerService) {
+    this.bindListener();
+  }
+
+  async bindListener() {
+    AMQPService.eventEmitter.on(AMQP_CONNECTION_DISCONNECTED, () => {
+      this.logger.debug('received disconnected event');
+    });
+  }
 
   getHello(): string {
     return 'Hello World!';