feat(messaging): provide info about timeout and transport type
MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -13,17 +13,22 @@ class AmqpStrategyConnection implements TransportLayerConnection {
private closeSubject$ = new Subject();
constructor(
- isConsumer: boolean,
+ public type: Transport,
+ private isConsumer: boolean,
private connectionManager: AmqpConnectionManager,
private channelWrapper: ChannelWrapper,
private options: AmqpStrategyOptions,
) {
process.nextTick(async () => {
- if (isConsumer) await this.consumeMessages();
+ if (this.isConsumer) await this.consumeMessages();
this.statusSubject$.next(AmqpConnectionStatus.CONNECTED);
});
}
+ get config() {
+ return { timeout: this.options.timeout ?? DEFAULT_TIMEOUT };
+ }
+
get close$() {
return this.closeSubject$.asObservable();
}
@@ -149,14 +154,16 @@ class AmqpStrategyConnection implements TransportLayerConnection {
};
ackMessage = (msg: TransportMessage | undefined) => {
- if (msg) {
+ if (this.options.expectAck && msg && !msg.raw.isAcked) {
this.channelWrapper.ack(msg.raw);
+ msg.raw.isAcked = true;
}
}
nackMessage = (msg: TransportMessage | undefined, resend = true) => {
- if (msg) {
+ if (this.options.expectAck && msg && !msg.raw.isNacked) {
this.channelWrapper.nack(msg.raw, false , resend);
+ msg.raw.isNacked = true;
}
}
@@ -207,6 +214,7 @@ class AmqpStrategy implements TransportLayer {
await (channelWrapper as any).waitForConnect(); // not available in @types/amqp-connection-manager
return new AmqpStrategyConnection(
+ Transport.AMQP,
isConsumer,
connectionManager,
channelWrapper,

MODIFY
packages/messaging/src/transport/strategies/local.strategy.ts
packages/messaging/src/transport/strategies/local.strategy.ts
@@ -11,7 +11,14 @@ class LocalStrategyConnection implements TransportLayerConnection {
private msgSubject$ = new Subject<TransportMessage<Buffer>>();
private closeSubject$ = new Subject();
- constructor(private opts: TransportLayerConfig) { }
+ constructor(
+ public type: Transport,
+ private opts: TransportLayerConfig,
+ ) { }
+
+ get config() {
+ return { timeout: this.opts.timeout };
+ }
get status$() {
return EMPTY;
@@ -88,7 +95,10 @@ class LocalStrategy implements TransportLayer {
}
async connect() {
- return new LocalStrategyConnection(this.config);
+ return new LocalStrategyConnection(
+ Transport.LOCAL,
+ this.config,
+ );
}
}

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -19,6 +19,7 @@ class RedisStrategyConnection implements TransportLayerConnection {
private closeSubject$ = new Subject();
constructor(
+ public type: Transport,
private opts: { channel: string; isConsumer: boolean; timeout: number },
private publisher: RedisClient,
private subscriber: RedisClient,
@@ -33,6 +34,10 @@ class RedisStrategyConnection implements TransportLayerConnection {
process.nextTick(() => this.statusSubject$.next(RedisConnectionStatus.CONNECT));
}
+ get config() {
+ return { timeout: this.opts.timeout };
+ }
+
get status$() {
const ready$ = merge(fromEvent(this.subscriber, 'ready'))
.pipe(mapTo(RedisConnectionStatus.READY));
@@ -140,12 +145,19 @@ class RedisStrategy implements TransportLayer {
const publisher = await RedisHelper.connectClient(this.clientOpts);
const subscriber = await RedisHelper.connectClient(this.clientOpts);
const rpcSubscriber = await RedisHelper.connectClient(this.clientOpts);
+ const opts = { isConsumer, channel, timeout: timeout ?? DEFAULT_TIMEOUT };
if (isConsumer) {
await RedisHelper.subscribeChannel(subscriber)(channel);
}
- return new RedisStrategyConnection({ isConsumer, channel, timeout: timeout ?? DEFAULT_TIMEOUT }, publisher, subscriber, rpcSubscriber);
+ return new RedisStrategyConnection(
+ Transport.REDIS,
+ opts,
+ publisher,
+ subscriber,
+ rpcSubscriber,
+ );
}
}

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -43,6 +43,8 @@ export interface TransportLayerConnection {
nackMessage: (message: TransportMessage | undefined, resend?: boolean) => void;
close: () => Promise<any>;
getChannel(): string;
+ type: Transport;
+ config: { timeout: number };
message$: Observable<TransportMessage<Buffer>>;
status$: Observable<string>;
error$: Observable<Error>;

feat(messaging): provide info about timeout and transport type