amirho/nest
1
0
Fork 0
mirror of https://github.com/nestjs/nest.git synced 2026-02-21 23:11:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(microservices): add max connections attempts to rmq, hide urls

Browse Source
This commit is contained in:
Kamil Myśliwiec
2022-07-28 14:34:02 +02:00
parent 7edb26e429
commit ddbba920cd
2 changed files with 37 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
packages/microservices/interfaces/microservice-configuration.interface.ts
View File

@@ -183,6 +183,13 @@ export interface RmqOptions {
persistent?: boolean;
headers?: Record<string, string>;
noAssert?: boolean;
/**
* Maximum number of connection attempts.
* Applies only to the consumer configuration.
* -1 === infinite
* @default -1
*/
maxConnectionAttempts?: number;
};
}

36
packages/microservices/server/server-rmq.ts
View File

@@ -4,15 +4,15 @@ import {
isUndefined,
} from '@nestjs/common/utils/shared.utils';
import {
CONNECTION_FAILED_MESSAGE,
CONNECT_EVENT,
CONNECT_FAILED_EVENT,
CONNECTION_FAILED_MESSAGE,
DISCONNECT_EVENT,
DISCONNECTED_RMQ_MESSAGE,
DISCONNECT_EVENT,
NO_MESSAGE_HANDLER,
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT,
RQM_DEFAULT_NO_ASSERT,
RQM_DEFAULT_NOACK,
RQM_DEFAULT_NO_ASSERT,
RQM_DEFAULT_PREFETCH_COUNT,
RQM_DEFAULT_QUEUE,
RQM_DEFAULT_QUEUE_OPTIONS,
@@ -31,11 +31,14 @@ import { Server } from './server';
let rqmPackage: any = {};
const INFINITE_CONNECTION_ATTEMPTS = -1;
export class ServerRMQ extends Server implements CustomTransportStrategy {
public readonly transportId = Transport.RMQ;
protected server: any = null;
protected channel: any = null;
protected connectionAttempts = 0;
protected readonly urls: string[] | RmqUrl[];
protected readonly queue: string;
protected readonly prefetchCount: number;
@@ -86,7 +89,9 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
this.server && this.server.close();
}
public async start(callback?: () => void) {
public async start(
callback?: (err?: unknown, ...optionalParams: unknown[]) => void,
) {
this.server = this.createClient();
this.server.on(CONNECT_EVENT, () => {
if (this.channel) {
@@ -97,13 +102,32 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
setup: (channel: any) => this.setupChannel(channel, callback),
});
});
const maxConnectionAttempts = this.getOptionsProp(
this.options,
'maxConnectionAttempts',
INFINITE_CONNECTION_ATTEMPTS,
);
this.server.on(DISCONNECT_EVENT, (err: any) => {
this.logger.error(DISCONNECTED_RMQ_MESSAGE);
this.logger.error(err);
});
this.server.on(CONNECT_FAILED_EVENT, (err: any) => {
this.server.on(CONNECT_FAILED_EVENT, (error: Record<string, unknown>) => {
this.logger.error(CONNECTION_FAILED_MESSAGE);
this.logger.error(err);
if (error?.err) {
this.logger.error(error.err);
}
const isReconnecting = !!this.channel;
if (
maxConnectionAttempts === INFINITE_CONNECTION_ATTEMPTS ||
isReconnecting
) {
return;
}
if (++this.connectionAttempts === maxConnectionAttempts) {
this.close();
callback?.(error.err ?? new Error(CONNECTION_FAILED_MESSAGE));
}
});
}