fix(messaging): high spike of RPC queues for AMQP strategy
MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
+import { pipe } from 'fp-ts/lib/function';
import { Subject, fromEvent, merge, from } from 'rxjs';
import { map, filter, take, mapTo, first, mergeMap, share, tap } from 'rxjs/operators';
import { Channel, ConsumeMessage, Replies } from 'amqplib';
@@ -109,7 +110,7 @@ class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP>
});
});
- await this.channelWrapper.addSetup(async (channel: Channel) => {
+ const modifyChannelSetup = async (channel: Channel): Promise<void> => {
const replyQueue = await channel.assertQueue('', {
exclusive: true,
autoDelete: true,
@@ -122,19 +123,24 @@ class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP>
{ noAck: true },
);
- setTimeout(() => {
- channel
- .cancel(consumer.consumerTag)
- .catch(error => resSubject$.error(error));
- }, timeout);
+ setTimeout(async () => await removeChannelSetupForConsumerTag(consumer.consumerTag), timeout);
replyToSubject.next(replyQueue.queue);
- });
+ };
+
+ const removeChannelSetupForConsumerTag = (consumerTag: string): Promise<void> =>
+ this.channelWrapper
+ .removeSetup(modifyChannelSetup, async (channel: Channel) => channel.cancel(consumerTag))
+ .catch(error => resSubject$.error(error));
+
+ await this.channelWrapper.addSetup(modifyChannelSetup);
return resSubject$.asObservable().pipe(
filter(raw => raw.msg.properties.correlationId === correlationId),
take(1),
- mergeMap(raw => from(this.channelWrapper.addSetup((channel: Channel) => channel.cancel(raw.tag))).pipe(
+ mergeMap(raw =>
+ pipe(
+ from(removeChannelSetupForConsumerTag(raw.tag)),
mapTo(({
data: raw.msg.content,
replyTo: raw.msg.properties.replyTo,

fix(messaging): high spike of RPC queues for AMQP strategy