imp(messaging): auto-delete replyQueue
MODIFY
packages/messaging/src/client/messaging.client.interface.ts
packages/messaging/src/client/messaging.client.interface.ts
@@ -7,7 +7,6 @@ import { TransportMessageTransformer } from '../transport/transport.interface';
export interface MessagingClient {
emit: <T>(data: T) => Observable<any>;
send: <T, U>(data: T) => Observable<U>;
- publish: <T>(data: T) => Observable<any>;
close: () => Observable<any>;
}

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -40,7 +40,10 @@ class AmqpStrategyConnection implements TransportLayerConnection {
sendMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
const { correlationId, data } = msg;
const resSubject$ = new Subject<ConsumeMessage>();
- const replyQueue = await this.channel.assertQueue('', { exclusive: true });
+ const replyQueue = await this.channel.assertQueue('', {
+ exclusive: true,
+ autoDelete: true,
+ });
const consumer = await this.channel.consume(
replyQueue.queue,
@@ -48,7 +51,10 @@ class AmqpStrategyConnection implements TransportLayerConnection {
{ noAck: true },
);
- this.channel.sendToQueue(queue, data, { correlationId, replyTo: replyQueue.queue });
+ this.channel.sendToQueue(queue, data, {
+ correlationId,
+ replyTo: replyQueue.queue,
+ });
return resSubject$.asObservable().pipe(
filter(raw => raw.properties.correlationId === correlationId),

MODIFY
packages/messaging/src/transport/transport.provider.ts
packages/messaging/src/transport/transport.provider.ts
import { Transport, TransportLayer } from './transport.interface';
import { createAmqpStrategy } from './strategies/amqp.strategy';
-// @TODO
-import { createTcpStrategy } from './strategies/tcp.strategy';
-import { createNatsStrategy } from './strategies/nats.strategy';
+// @TODO // import { createTcpStrategy } from './strategies/tcp.strategy';
+// @TODO // import { createNatsStrategy } from './strategies/nats.strategy';
export const provideTransportLayer = (transport: Transport, transportOptions: any): TransportLayer => {
switch (transport) {
case Transport.AMQP:
return createAmqpStrategy(transportOptions);
default:
- // @TODO
- return createAmqpStrategy(transportOptions);
+ throw new Error('Unsupported transport type');
}
};

imp(messaging): auto-delete replyQueue