feature - reconnect fix and async init

piotrfrankowski · Apr 24, 2020 · 88ed201 · 88ed201
1 parent 33aa980
commit 88ed201
Show file tree

Hide file tree

Showing 6 changed files with 60 additions and 50 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,11 +1,15 @@
 # Change log
 
-## 1.5.2
--   Fixed double logging
+## 1.7.0
+-   Fixed reconnection bug
+-   Async init all modules loaded (thx to mjarmoc)
 
 ## 1.6.0
 -   Custom logger injection (thx to @minenkom)
 
+## 1.5.2
+-   Fixed double logging
+
 ## 1.5.1
 -   Fixed ack race condition
 -   Added tests

diff --git a/e2e/tests/rmq.e2e-spec.ts b/e2e/tests/rmq.e2e-spec.ts
@@ -32,7 +32,7 @@ describe('RMQe2e', () => {
 					middleware: [DoublePipe],
 					intercepters: [ZeroIntercepter],
 					errorHandler: ErrorHostHandler,
-					serviceName: 'test-service'
+					serviceName: 'test-service',
 				}),
 			],
 			controllers: [ApiController, MicroserviceController],
@@ -68,7 +68,7 @@ describe('RMQe2e', () => {
 		});
 		it('get common Error from method', async () => {
 			try {
-				const { result } = await apiController.sumSuccess([0,0,0]);
+				const { result } = await apiController.sumSuccess([0, 0, 0]);
 				expect(result).not.toBe(0);
 			} catch (error) {
 				expect(error.message).toBe('My error from method');
@@ -81,7 +81,7 @@ describe('RMQe2e', () => {
 		});
 		it('get RMQError from method', async () => {
 			try {
-				const { result } = await apiController.sumSuccess([-1,0,0]);
+				const { result } = await apiController.sumSuccess([-1, 0, 0]);
 				expect(result).not.toBe(-1);
 			} catch (error) {
 				expect(error.message).toBe('My RMQError from method');
@@ -94,7 +94,7 @@ describe('RMQe2e', () => {
 		});
 		it('get undefined return Error', async () => {
 			try {
-				const { result } = await apiController.sumSuccess([-11,0,0]);
+				const { result } = await apiController.sumSuccess([-11, 0, 0]);
 				expect(result).not.toBe(-11);
 			} catch (error) {
 				expect(error.message).toBe(ERROR_UNDEFINED_FROM_RPC);
@@ -138,7 +138,7 @@ describe('RMQe2e', () => {
 	describe('errorHandler', () => {
 		it('error host change', async () => {
 			try {
-				const { result } = await apiController.sumSuccess([0,0,0]);
+				const { result } = await apiController.sumSuccess([0, 0, 0]);
 				expect(result).not.toBe(0);
 			} catch (error) {
 				expect(error.host).toBe('handler');

diff --git a/lib/rmq.module.ts b/lib/rmq.module.ts
@@ -9,7 +9,9 @@ export class RMQModule {
 		const rmqServiceProvider = {
 			provide: RMQService,
 			useFactory: async (): Promise<RMQService> => {
-				return new RMQService(options);
+				const RMQInstance = new RMQService(options);
+				await RMQInstance.init();
+				return RMQInstance;
 			},
 		};
 		return {

diff --git a/lib/rmq.service.ts b/lib/rmq.service.ts
@@ -41,45 +41,49 @@ export class RMQService {
 	constructor(options: IRMQServiceOptions) {
 		this.options = options;
 		this.logger = options.logger ? options.logger : new RQMColorLogger(this.options.logMessages);
-		this.init();
 	}
 
 	public async init(): Promise<void> {
-		const connectionURLs: string[] = this.options.connections.map((connection: IRMQConnection) => {
-			return `amqp://${connection.login}:${connection.password}@${connection.host}`;
-		});
-		const connectionOptions = {
-			reconnectTimeInSeconds: this.options.reconnectTimeInSeconds ?? DEFAULT_RECONNECT_TIME
-		};
-		this.server = amqp.connect(connectionURLs, connectionOptions);
-		this.channel = this.server.createChannel({
-			json: false,
-			setup: async (channel: Channel) => {
-				await channel.assertExchange(this.options.exchangeName, EXCHANGE_TYPE, {
-					durable: this.options.isExchangeDurable ?? true,
-				});
-				await channel.prefetch(
-					this.options.prefetchCount ?? DEFAULT_PREFETCH_COUNT,
-					this.options.isGlobalPrefetchCount ?? false
-				);
-				await channel.consume(
-					this.replyQueue,
-					(msg: Message) => {
-						this.sendResponseEmitter.emit(msg.properties.correlationId, msg);
-					},
-					{ noAck: true }
-				);
-				this.waitForReply();
-				if (this.options.queueName) {
-					this.listen(channel);
-				}
-				this.logger.log(CONNECTED_MESSAGE);
-			},
-		});
+		return new Promise((resolve => {
+			const connectionURLs: string[] = this.options.connections.map((connection: IRMQConnection) => {
+				return `amqp://${connection.login}:${connection.password}@${connection.host}`;
+			});
+			const connectionOptions = {
+				reconnectTimeInSeconds: this.options.reconnectTimeInSeconds ?? DEFAULT_RECONNECT_TIME
+			};
+			this.server = amqp.connect(connectionURLs, connectionOptions);
+			this.channel = this.server.createChannel({
+				json: false,
+				setup: async (channel: Channel) => {
+					await channel.assertExchange(this.options.exchangeName, EXCHANGE_TYPE, {
+						durable: this.options.isExchangeDurable ?? true,
+					});
+					await channel.prefetch(
+						this.options.prefetchCount ?? DEFAULT_PREFETCH_COUNT,
+						this.options.isGlobalPrefetchCount ?? false
+					);
+					await channel.consume(
+						this.replyQueue,
+						(msg: Message) => {
+							this.sendResponseEmitter.emit(msg.properties.correlationId, msg);
+						},
+						{ noAck: true }
+					);
+					this.waitForReply();
+					if (this.options.queueName) {
+						this.listen(channel);
+					}
+					this.logger.log(CONNECTED_MESSAGE);
+					resolve();
+				},
+			});
+
+			this.server.on(DISCONNECT_EVENT, err => {
+				this.logger.error(DISCONNECT_MESSAGE);
+				this.logger.error(err.err);
+			});
+		}));
 
-		this.server.on(DISCONNECT_EVENT, err => {
-			this.logger.error(DISCONNECT_MESSAGE, err);
-		});
 	}
 
 	public async send<IMessage, IReply>(topic: string, message: IMessage): Promise<IReply> {
@@ -150,7 +154,7 @@ export class RMQService {
 		);
 	}
 
-	private async waitForReply(): Promise<void> {
+	private waitForReply(): void {
 		responseEmitter.on(ResponseEmmiterResult.success, async (msg, result) => {
 			this.reply(result, msg);
 		});

diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
 	"name": "nestjs-rmq",
-	"version": "1.6.0",
+	"version": "1.7.0",
 	"description": "NestJS RabbitMQ Module",
 	"main": "./dist/index.js",
 	"types": "./dist/index.d.ts",
@@ -37,7 +37,7 @@
 		"@nestjs/core": "^7.0.0",
 		"@nestjs/platform-express": "^7.0.0",
 		"@nestjs/testing": "^7.0.0",
-		"@types/amqp-connection-manager": "^2.0.8",
+		"@types/amqp-connection-manager": "^2.0.9",
 		"@types/amqplib": "^0.5.13",
 		"@types/chalk": "^2.2.0",
 		"@types/jest": "^25.1.4",