message: fix(messaging): high spike of RPC queues for AMQP strategy
diff: +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,
  