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

fix(microservices): Prevent unusable queue assertion

Browse Source 
The client queue assertion and bind prevent for cases when exchange will be used for send
This commit is contained in:
Serafim Gerasimov
2025-10-08 23:53:23 +03:00
parent bf59ad2ecc
commit c58da76adf
2 changed files with 36 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -204,22 +204,19 @@ export class ClientRMQ extends ClientProxy<RmqEvents, RmqStatus> {
this.getOptionsProp(this.options, 'isGlobalPrefetchCount') ||
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
if (!this.noAssert) {
await channel.assertQueue(this.queue, this.queueOptions);
}
if (!this.options.wildcards && this.options.exchangeType !== 'fanout') {
if (!this.noAssert) {
await channel.assertQueue(this.queue, this.queueOptions);
}
if (
this.options.exchange &&
(this.options.routingKey || this.options.exchangeType === 'fanout')
) {
await channel.bindQueue(
this.queue,
this.options.exchange,
this.options.exchangeType === 'fanout' ? '' : this.options.routingKey,
);
}
if (this.options.wildcards) {
if (this.options.exchange && this.options.routingKey) {
await channel.bindQueue(
this.queue,
this.options.exchange,
this.options.exchangeType === 'fanout' ? '' : this.options.routingKey,
);
}
} else {
const exchange = this.getOptionsProp(
this.options,
'exchange',

33
packages/microservices/test/client/client-rmq.spec.ts
View File

@@ -125,7 +125,6 @@ describe('ClientRMQ', function () {
const prefetchCount = 10;
let consumeStub: sinon.SinonStub;
let bindQueueStub: sinon.SinonStub;
let channel: any = {};
beforeEach(() => {
@@ -136,9 +135,9 @@ describe('ClientRMQ', function () {
channel = {
assertQueue: sinon.spy(() => ({})),
prefetch: sinon.spy(),
bindQueue: bindQueueStub,
bindQueue: sinon.spy(),
assertExchange: sinon.spy(),
};
bindQueueStub = sinon.stub();
consumeStub = sinon.stub(client, 'consumeChannel').callsFake(() => null!);
});
afterEach(() => {
@@ -150,18 +149,34 @@ describe('ClientRMQ', function () {
await client.setupChannel(channel, () => null);
expect(channel.assertQueue.calledWith(queue, queueOptions)).to.be.true;
});
it('should call "bindQueue" with exchangeType is fanout', async () => {
untypedClient['options']['exchangeType'] = 'fanout';
untypedClient['options']['exchange'] = exchange;
await client.setupChannel(channel, () => null);
expect(channel.bindQueue.calledWith(queue, exchange, '')).to.be.true;
});
it('should not call "assertQueue" when noAssert is true', async () => {
client['noAssert'] = true;
await client.setupChannel(channel, () => null);
expect(channel.assertQueue.called).not.to.be.true;
});
it('should not call "assertQueue" when exchangeType is fanout', async () => {
untypedClient['options']['exchangeType'] = 'fanout';
untypedClient['options']['exchange'] = exchange;
await client.setupChannel(channel, () => null);
expect(channel.assertQueue.called).not.to.be.true;
});
it('should not call "assertQueue" when wildcards is true', async () => {
untypedClient['options']['wildcards'] = true;
await client.setupChannel(channel, () => null);
expect(channel.assertQueue.called).not.to.be.true;
});
it('should not call "bindQueue" when exchangeType is fanout', async () => {
untypedClient['options']['exchangeType'] = 'fanout';
untypedClient['options']['exchange'] = exchange;
await client.setupChannel(channel, () => null);
expect(channel.bindQueue.called).not.to.be.true;
});
it('should not call "bindQueue" when wildcards is true', async () => {
untypedClient['options']['wildcards'] = true;
await client.setupChannel(channel, () => null);
expect(channel.bindQueue.called).not.to.be.true;
});
it('should call "prefetch" with prefetchCount and "isGlobalPrefetchCount"', async () => {
await client.setupChannel(channel, () => null);
expect(channel.prefetch.calledWith(prefetchCount, isGlobalPrefetchCount))