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

Merge pull request #9798 from frankmangone/frankmangone/feature/avoid-queue-declaration-rmq

Browse Source 
feat(microservices): add noAssert option for RMQ connection
This commit is contained in:
Kamil Mysliwiec
2022-07-20 10:40:26 +02:00
committed by GitHub
parent 429dfa1382 bb7e9d7837
commit 566301232b
4 changed files with 17 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
packages/microservices/client/client-rmq.ts
View File

@@ -16,6 +16,7 @@ import {
RQM_DEFAULT_QUEUE,
RQM_DEFAULT_QUEUE_OPTIONS,
RQM_DEFAULT_URL,
RQM_DEFAULT_NO_ASSERT,
} from '../constants';
import { RmqUrl } from '../external/rmq-url.interface';
import { ReadPacket, RmqOptions, WritePacket } from '../interfaces';
@@ -38,6 +39,7 @@ export class ClientRMQ extends ClientProxy {
protected responseEmitter: EventEmitter;
protected replyQueue: string;
protected persistent: boolean;
protected noAssert: boolean;
constructor(protected readonly options: RmqOptions['options']) {
super();
@@ -51,6 +53,9 @@ export class ClientRMQ extends ClientProxy {
this.getOptionsProp(this.options, 'replyQueue') || REPLY_QUEUE;
this.persistent =
this.getOptionsProp(this.options, 'persistent') || RQM_DEFAULT_PERSISTENT;
this.noAssert =
this.getOptionsProp(this.options, 'noAssert') || RQM_DEFAULT_NO_ASSERT;
loadPackage('amqplib', ClientRMQ.name, () => require('amqplib'));
rqmPackage = loadPackage('amqp-connection-manager', ClientRMQ.name, () =>
require('amqp-connection-manager'),
@@ -143,7 +148,9 @@ export class ClientRMQ extends ClientProxy {
this.getOptionsProp(this.options, 'isGlobalPrefetchCount') ||
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
await channel.assertQueue(this.queue, this.queueOptions);
if (!this.queueOptions.noAssert) {
await channel.assertQueue(this.queue, this.queueOptions);
}
await channel.prefetch(prefetchCount, isGlobalPrefetchCount);
this.responseEmitter = new EventEmitter();

1
packages/microservices/constants.ts
View File

@@ -39,6 +39,7 @@ export const RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT = false;
export const RQM_DEFAULT_QUEUE_OPTIONS = {};
export const RQM_DEFAULT_NOACK = true;
export const RQM_DEFAULT_PERSISTENT = false;
export const RQM_DEFAULT_NO_ASSERT = false;
export const GRPC_DEFAULT_PROTO_LOADER = '@grpc/proto-loader';
export const NO_EVENT_HANDLER = (text: TemplateStringsArray, pattern: string) =>

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

@@ -181,6 +181,7 @@ export interface RmqOptions {
replyQueue?: string;
persistent?: boolean;
headers?: Record<string, string>;
noAssert?: boolean;
};
}

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

@@ -16,6 +16,7 @@ import {
RQM_DEFAULT_QUEUE,
RQM_DEFAULT_QUEUE_OPTIONS,
RQM_DEFAULT_URL,
RQM_DEFAULT_NO_ASSERT,
} from '../constants';
import { RmqContext } from '../ctx-host';
import { Transport } from '../enums';
@@ -40,6 +41,7 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
protected readonly prefetchCount: number;
protected readonly queueOptions: any;
protected readonly isGlobalPrefetchCount: boolean;
protected readonly noAssert: boolean;
constructor(protected readonly options: RmqOptions['options']) {
super();
@@ -55,6 +57,8 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
this.queueOptions =
this.getOptionsProp(this.options, 'queueOptions') ||
RQM_DEFAULT_QUEUE_OPTIONS;
this.noAssert =
this.getOptionsProp(this.options, 'noAssert') || RQM_DEFAULT_NO_ASSERT;
this.loadPackage('amqplib', ServerRMQ.name, () => require('amqplib'));
rqmPackage = this.loadPackage(
@@ -115,7 +119,9 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
public async setupChannel(channel: any, callback: Function) {
const noAck = this.getOptionsProp(this.options, 'noAck', RQM_DEFAULT_NOACK);
await channel.assertQueue(this.queue, this.queueOptions);
if (!this.queueOptions.noAssert) {
await channel.assertQueue(this.queue, this.queueOptions);
}
await channel.prefetch(this.prefetchCount, this.isGlobalPrefetchCount);
channel.consume(
this.queue,