Merge pull request #35 from hapinessjs/next

Next into master

Author: fauvillej

README.md

@@ -453,6 +453,12 @@ To set up your development environment:
 [Back to top](#table-of-contents)
 
 ## Change History
+* v1.4.2 (2018-06-11)
+    * Do not retry to connect if closing server
+* v1.4.1 (2018-05-31)
+    * Fix channel creation after reconnection
+* v1.4.0 (2018-04-24)
+    * Refactor channel management to handle connection errors
 * v1.3.0 (2018-03-27)
     * Add shutdown (SIGTERM/SIGINT) support
 * v1.2.3 (2018-02-05)

package-lock.json

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

package.json

@@ -19,7 +19,7 @@ export class ConnectionManager extends EventEmitter {
     private _connect: typeof connect;
     private _defaultPrefetch: number;
     private _channelStore: ChannelStore;
-    private _isSIGTERMReceived: boolean;
+    private _closingServer: boolean;
 
     constructor(config?: RabbitMQConfigConnection) {
         super();
@@ -28,6 +28,7 @@ export class ConnectionManager extends EventEmitter {
         this._connection = null;
         this._isConnecting = false;
         this._isConnected = false;
+        this._closingServer = false;
         this._options = Object.assign({}, config);
         this._options.retry = Object.assign({ delay: 5000, maximum_attempts: -1 }, this._options.retry);
 
@@ -48,9 +49,9 @@ export class ConnectionManager extends EventEmitter {
 
         // Will block new connection if SIGTERM is received
         /* istanbul ignore next */
-        process.once('SIGTERM', () => this._isSIGTERMReceived = true);
+        process.once('SIGTERM', () => this._closingServer = true);
         /* istanbul ignore next */
-        process.once('SIGINT', () => this._isSIGTERMReceived = true);
+        process.once('SIGINT', () => this._closingServer = true);
 
         this.setDefaultPrefetch(this._options.default_prefetch);
 
@@ -88,13 +89,15 @@ export class ConnectionManager extends EventEmitter {
         return Observable.of(null)
             .flatMap(() => {
                 debug('try to open connection ...');
-                debug(this._options.retry.delay);
+                debug(`Retry delay: ${this._options.retry.delay}`);
                 return Observable.fromPromise(this._connect(this._uri));
             })
             .retryWhen(errors => {
                 errors.forEach(err => debug(err.message, err.stack));
                 return errors
+                    .scan(attempts => attempts + 1, 0)
                     .delay(this._options.retry.delay)
+                    .takeWhile((attempts) => attempts < this._options.retry.maximum_attempts && !this._closingServer)
                     .take(this._options.retry.maximum_attempts)
                     .concat(Observable.throw(new Error('Retry limit exceeded')))
             });
@@ -105,11 +108,8 @@ export class ConnectionManager extends EventEmitter {
             return Observable.of(null);
         }
 
-        /* istanbul ignore next */
-        if (this._isSIGTERMReceived) {
-            return Observable.of(null);
-        }
 
+        this._closingServer = false;
         this._isConnecting = true;
 
         debug('Connecting', this._uri);
@@ -138,6 +138,7 @@ export class ConnectionManager extends EventEmitter {
 
     close(): Observable<void> {
         this._isConnected = false;
+        this._closingServer = true;
         return Observable.fromPromise(this._connection.close());
     }
 

src/module/managers/connection-manager.ts

@@ -22,15 +22,20 @@ export class InitExtensionUnitTest {
     private queue: QueueManager;
     private userQueue;
 
-    before() {
-        const connection = new ConnectionManagerMock();
-        this.ch = new ChannelManager(connection);
-        this.ch['ch'] = <any>new ChannelMock();
-        this.userQueue = new UserQueue();
-        this.queueWrapper = new QueueWrapper(this.userQueue, extractMetadataByDecorator(UserQueue, 'Queue'));
-        this.messageRouter = new MessageRouter();
-        this.queue = new QueueManager(this.ch, this.queueWrapper);
-        unit.spy(this.userQueue, 'onMessage');
+    before(done) {
+        try {
+            const connection = new ConnectionManagerMock();
+            this.ch = new ChannelManager(connection);
+            this.ch['ch'] = <any>new ChannelMock();
+            this.userQueue = new UserQueue();
+            this.queueWrapper = new QueueWrapper(this.userQueue, extractMetadataByDecorator(UserQueue, 'Queue'));
+            this.messageRouter = new MessageRouter();
+            this.queue = new QueueManager(this.ch, this.queueWrapper);
+            unit.spy(this.userQueue, 'onMessage');
+            done();
+        } catch (err) {
+            done(err);
+        }
     }
 
     after() {

test/unit/extension/init-extension.test.ts

@@ -97,7 +97,9 @@ export class ConnectionUnitTest {
                 unit.number(connectStub.callCount).is(3);
                 done();
             },
-            () => done(new Error('Should not be here'))
+            (err) => {
+                done(new Error('Should not be here'));
+            }
         );
     }
 

test/unit/managers/connection.test.ts

@@ -81,9 +81,11 @@ export class MessageStoreTest {
         MessageStore['shutdownTimeoutMs'] = 500;
         MessageStore.addConsumer(<any>this.ch, 'consumer-1');
         MessageStore.addMessage(<any>{ uuid: 4 });
-        MessageStore.shutdown(this.connection).subscribe(() => done(new Error('Cannot succeed')), err => {
-            setTimeout(() => done(), 500);
-        });
+        MessageStore.shutdown(this.connection)
+            .subscribe(() => done(new Error('Cannot succeed')), err => {
+                setTimeout(() => done(), 500);
+            });
+
         setTimeout(() => {
             MessageStore.remove(<any>{ uuid: 4 });
         }, 510);