fix(channel closed): If channel is closed and we try to send a messag…

…e, emit an error on the connection and throw the error

.nvmrc

@@ -1 +1 @@
-7.10.0
+10

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@hapiness/rabbitmq",
-  "version": "1.7.0",
+  "version": "1.7.1",
   "description": "Hapiness module for rabbitmq",
   "main": "commonjs/index.js",
   "types": "index.d.ts",

src/module/message.ts

@@ -3,9 +3,27 @@ import * as _has from 'lodash.has';
 import * as _pick from 'lodash.pick';
 import { MessageOptions } from './interfaces';
 import { events } from './events';
+import { ConnectionManager } from './managers';
+
+function handleChannelClosedError(cb, ...options) {
+    try {
+        cb(...options);
+    } catch (err) {
+        if (err.message && err.message.toLowerCase().includes('channel closed')) {
+            // Add the code for this error, so it can be handled later
+            (<any>err).code = 'CHANNEL_CLOSED_ERROR';
+
+            // Emit the error on the connection object
+            (<ConnectionManager>(events.connection)).emit('error', err);
+        }
+
+        throw err;
+    }
+}
 
 export function sendMessage(ch: ChannelInterface, message: any, options: MessageOptions = {}): boolean {
     let { json } = Object.assign({ json: true }, options);
+
     const allowedOptions = [
         'expiration',
         'userId',
@@ -25,6 +43,7 @@ export function sendMessage(ch: ChannelInterface, message: any, options: Message
         'type',
         'appId'
     ];
+
     const publishOptions = _pick(options, allowedOptions);
 
     if (!ch) {
@@ -62,15 +81,28 @@ export function sendMessage(ch: ChannelInterface, message: any, options: Message
             queue: options.queue,
             content: message
         });
-        return ch.sendToQueue(options.queue, encodedMessage, publishOptions);
+
+        return handleChannelClosedError(
+            ch.sendToQueue,
+            options.queue,
+            encodedMessage,
+            publishOptions
+        );
     } else if (options.exchange) {
         events.message.emit('sent', {
             publishOptions,
             exchange: options.exchange,
             routingKey: options.routingKey,
             content: message
         });
-        return ch.publish(options.exchange, options.routingKey, encodedMessage, publishOptions);
+
+        return handleChannelClosedError(
+            ch.publish,
+            options.exchange,
+            options.routingKey,
+            encodedMessage,
+            publishOptions
+        );
     } else {
         throw new Error('Specify a queue or an exchange');
     }

src/module/services/message.service.ts

@@ -7,7 +7,9 @@ import { Channel } from 'amqplib';
 
 @Injectable()
 export class MessageService {
+
     private _sendMessage;
+
     constructor(private _channelService: ChannelService) {
         this._sendMessage = sendMessage;
     }