Merge pull request #31 from hapinessjs/next

release(version): v1.4.0
hapinessjs · Apr 24, 2018 · dc9cbd2 · dc9cbd2
2 parents aa935e1 + a97aa8a
commit dc9cbd2
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 5 deletions.
diff --git a/src/module/managers/channel-manager.ts b/src/module/managers/channel-manager.ts
@@ -32,7 +32,7 @@ export class ChannelManager extends EventEmitter {
         return this._global;
     }
 
-    public canCreateChannel() {
+    public canCreateChannel(): boolean {
         return this._connectionManager.isConnected() && !this._connectionManager.isConnecting() && !MessageStore.isShutdownRunning();
     }
 

diff --git a/src/module/managers/connection-manager.ts b/src/module/managers/connection-manager.ts
@@ -21,6 +21,7 @@ export class ConnectionManager extends EventEmitter {
     private _connect: typeof connect;
     private _defaultPrefetch: number;
     private _channelStore: ChannelStore;
+    private _isSIGTERMReceived: boolean;
 
     constructor(config?: RabbitMQConfigConnection) {
         super();
@@ -51,6 +52,10 @@ export class ConnectionManager extends EventEmitter {
             this._uri = `amqp://${credentials}${host}:${port}${vhost}${params}`;
         }
 
+        // Will block new connection if SIGTERM is received
+        process.once('SIGTERM', () => this._isSIGTERMReceived = true);
+        process.once('SIGINT', () => this._isSIGTERMReceived = true);
+
         this.setDefaultPrefetch(this._options.default_prefetch);
 
         // Create a channel store for this connection
@@ -104,6 +109,10 @@ export class ConnectionManager extends EventEmitter {
             return Observable.of(null);
         }
 
+        if (this._isSIGTERMReceived) {
+            return Observable.of(null);
+        }
+
         this._isConnecting = true;
 
         debug('Connecting', this._uri);

diff --git a/src/module/managers/message-store.ts b/src/module/managers/message-store.ts
@@ -4,6 +4,7 @@ import { Observable } from 'rxjs';
 import { Channel as ChannelInterface } from 'amqplib';
 import { RabbitMessage } from '../interfaces/rabbit-message';
 import { Config } from '@hapiness/config';
+import { ConnectionManager } from './connection-manager';
 
 const debug = require('debug')('hapiness:rabbitmq');
 
@@ -78,7 +79,7 @@ export class MessageStoreClass extends EventEmitter {
         this.shutdown_running = false;
     }
 
-    shutdown(connection): Observable<any> {
+    shutdown(connection: ConnectionManager): Observable<any> {
         if (this.shutdown_running) {
             debug('shutdown already running');
             return Observable.of(null);

diff --git a/src/module/managers/queue-manager.ts b/src/module/managers/queue-manager.ts
@@ -107,7 +107,7 @@ export class QueueManager {
         dispatcher,
         options,
         defaultDispatch
-    }) {
+    }): Observable<Replies.Consume> {
         const consumerChannel = this._ch.getChannel();
         return Observable.fromPromise(
             this._ch.getChannel().consume(this.getName(), message => {
@@ -150,6 +150,8 @@ export class QueueManager {
         )
         .do(res => MessageStore.addConsumer(consumerChannel, res.consumerTag))
         .do(res => {
+            // If channel is closed, has en error or is reconnected the consumerTag is not valid
+            // and needs to be removed
             ['close', 'error', 'reconnected'].forEach(event => this._ch.once(event, () => {
                 debug('removing consumer', res.consumerTag);
                 MessageStore.removeConsumer(res.consumerTag);
@@ -161,8 +163,7 @@ export class QueueManager {
         message: RabbitMessage,
         { storeMessage, options, err, consumerChannel }:
         { storeMessage: StoreMessage,
-            // options: { errorHandler: (err: Error, message: RabbitMessage, ch: ChannelInterface) => {} },
-            options: any,
+            options: { errorHandler: (err: Error, message: RabbitMessage, ch: ChannelInterface) => {} },
             err: Error, consumerChannel: ChannelInterface }
     ) {
         if (MessageStore.isShutdownRunning()) {