Merge pull request #37 from hapinessjs/next

version(v1.4.3)

Author: akanass

README.md

@@ -453,6 +453,8 @@ To set up your development environment:
 [Back to top](#table-of-contents)
 
 ## Change History
+* v1.4.3 (2018-08-20)
+    * Emit RETRY_LIMIT_EXCEEDED error on ConnectionManager
 * v1.4.2 (2018-06-11)
     * Do not retry to connect if closing server
 * v1.4.1 (2018-05-31)

package-lock.json

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

package.json

@@ -8,6 +8,8 @@ import { ChannelStore } from './channel-store';
 import { ChannelManager } from './channel-manager';
 
 const debug = require('debug')('hapiness:rabbitmq');
+const retryLimitErr: any = new Error('Retry limit exceeded');
+retryLimitErr.code = 'RETRY_LIMIT_EXCEEDED';
 
 export class ConnectionManager extends EventEmitter {
     private _connection: Connection;
@@ -99,7 +101,7 @@ export class ConnectionManager extends EventEmitter {
                     .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')))
+                    .concat(Observable.throw(retryLimitErr))
             });
     }
 
@@ -108,7 +110,6 @@ export class ConnectionManager extends EventEmitter {
             return Observable.of(null);
         }
 
-
         this._closingServer = false;
         this._isConnecting = true;
 
@@ -117,6 +118,10 @@ export class ConnectionManager extends EventEmitter {
         this.emitEvent('connecting');
         const obs = this.openConnection();
         return obs
+            .catch(err => {
+                this.emitEvent('error', err);
+                return Observable.throw(err);
+            })
             .flatMap(con => {
                 debug('connected, creating default channel ...');
                 this._connection = con;

src/module/managers/connection-manager.ts

@@ -61,7 +61,9 @@ export class RabbitMQExt implements OnExtensionLoad, OnModuleInstantiated, OnShu
             connection
                 .connect()
                 .flatMap(() => RegisterAnnotations.bootstrap(module, connection))
-                .subscribe(() => {}, err => errorHandler(err));
+                .subscribe(() => {}, err => {
+                    errorHandler(err);
+                });
         });
 
         return RegisterAnnotations.bootstrap(module, connection);

src/module/rabbitmq.extension.ts

@@ -136,4 +136,23 @@ export class ConnectionUnitTest {
         instance.setDefaultPrefetch(5);
         unit.number(instance.getDefaultPrefetch()).is(5);
     }
+
+    @test(' - Test openConnection error')
+    testOpenConnectionError(done) {
+        const instance = new ConnectionManager({ retry: { delay: 100, maximum_attempts: 1 } });
+        instance['_connection'] = <any>new RabbitConnectionMock();
+        unit.stub(instance, '_connect').returns(Promise.reject(Observable.throw(new Error('Woopsie'))));
+
+        Observable.forkJoin([
+            Observable.fromEvent(instance, 'error').map(err => {
+                unit.object(err).isInstanceOf(Error).hasProperty('code', 'RETRY_LIMIT_EXCEEDED');
+            }),
+            instance.connect()
+                .catch(err => {
+                    unit.object(err).isInstanceOf(Error).hasProperty('code', 'RETRY_LIMIT_EXCEEDED');
+                    return Observable.throw(null);
+                })
+                .flatMap(() => Observable.throw(new Error('Cannot be here')))
+        ]).subscribe(() => done(), err => done(err));
+    }
 }