feat(messaging): timeout RPC events and cleanup AMQP resources
MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
@@ -46,6 +46,15 @@ const test$ = r.pipe(
)),
);
+const timeout$ = r.pipe(
+ r.matchPath('/timeout'),
+ r.matchType('GET'),
+ r.useEffect((req$, { ask }) => req$.pipe(
+ mergeMap(() => useContext(ClientToken)(ask).send({ type: 'TIMEOUT' })),
+ mapTo(({ body: 'OK' })),
+ )),
+);
+
const fib$ = r.pipe(
r.matchPath('/fib/:number'),
r.matchType('GET'),
@@ -78,7 +87,7 @@ export const server = createServer({
port,
httpListener: httpListener({
middlewares: [logger$()],
- effects: [test$, fib$],
+ effects: [test$, timeout$, fib$],
}),
dependencies: [
bindEagerlyTo(ClientToken)(client),

MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
import { of } from 'rxjs';
-import { map, tap, mergeMap, bufferCount, mapTo, catchError } from 'rxjs/operators';
+import { map, tap, mergeMap, bufferCount, mapTo, catchError, delay } from 'rxjs/operators';
import { matchEvent, use, Event } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { createMicroservice, messagingListener, Transport, MsgEffect } from '@marblejs/messaging';
@@ -26,6 +26,16 @@ const test$: MsgEffect = event$ =>
mapTo({ type: 'TEST_RESULT' }),
);
+
+const timeout$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('TIMEOUT'),
+ mergeMap(event => of(event).pipe(
+ delay(40 * 1000),
+ mapTo({ ...event, type: 'TIMEOUT_RESULT' } as Event),
+ )),
+ );
+
export const microservice = createMicroservice({
transport: Transport.AMQP,
options: {
@@ -34,7 +44,7 @@ export const microservice = createMicroservice({
queueOptions: { durable: false },
},
messagingListener: messagingListener({
- effects: [fibonacci$, test$],
+ effects: [fibonacci$, test$, timeout$],
}),
});

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -3,8 +3,8 @@ import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
import { reader, ServerEventStreamToken, matchEvent, ServerEvent, AllServerEvents } from '@marblejs/core';
import { from, Observable, EMPTY } from 'rxjs';
-import { mergeMap, take, map } from 'rxjs/operators';
-import { TransportMessage, TransportLayerConnection } from '../transport/transport.interface';
+import { mergeMap, take, map, timeout } from 'rxjs/operators';
+import { TransportMessage, TransportLayerConnection, TransportLayerConfig } from '../transport/transport.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
import { MessagingClient, MessagingClientConfig } from './messaging.client.interface';
@@ -23,8 +23,9 @@ export const messagingClient = (config: MessagingClientConfig) => {
);
}
- const send = (conn: TransportLayerConnection) => <T, U>(msg: T): Observable<U> =>
+ const send = (config: TransportLayerConfig, conn: TransportLayerConnection) => <T, U>(msg: T): Observable<U> =>
from(conn.sendMessage(conn.getChannel(), { data: msgTransformer.encode(msg as any) })).pipe(
+ timeout(config.timeout),
map(m => m as TransportMessage<Buffer>),
map(m => msgTransformer.decode(m.data) as U),
take(1),
@@ -53,7 +54,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
return {
emit: emit(connection),
- send: send(connection),
+ send: send(transportLayer.config, connection),
close: close(connection),
} as MessagingClient;
}));

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.interface.ts
packages/messaging/src/transport/strategies/amqp.strategy.interface.ts
@@ -12,6 +12,7 @@ export interface AmqpStrategyOptions {
queueOptions?: Options.AssertQueue;
prefetchCount?: number;
expectAck?: boolean;
+ timeout?: number;
}
export enum AmqpConnectionStatus {

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
import { Subject, fromEvent, merge, from } from 'rxjs';
import { map, filter, take, mapTo, first, mergeMap, share } from 'rxjs/operators';
-import { Channel, ConsumeMessage } from 'amqplib';
+import { Channel, ConsumeMessage, Replies } from 'amqplib';
import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
-import { TransportLayer, TransportMessage, TransportLayerConnection, Transport } from '../transport.interface';
+import { TransportLayer, TransportMessage, TransportLayerConnection, Transport, DEFAULT_TIMEOUT } from '../transport.interface';
import { AmqpStrategyOptions, AmqpConnectionStatus } from './amqp.strategy.interface';
class AmqpStrategyConnection implements TransportLayerConnection {
@@ -55,6 +55,7 @@ class AmqpStrategyConnection implements TransportLayerConnection {
sendMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
const { data } = msg;
const correlationId = createUuid();
+ const timeout = this.options.timeout ?? DEFAULT_TIMEOUT;
const replyToSubject = new Subject<string>();
const resSubject$ = new Subject<{ msg: ConsumeMessage; tag: string }>();
@@ -71,14 +72,21 @@ class AmqpStrategyConnection implements TransportLayerConnection {
const replyQueue = await channel.assertQueue('', {
exclusive: true,
autoDelete: true,
+ expires: timeout,
});
- const consumer = await channel.consume(
+ const consumer: Replies.Consume = await channel.consume(
replyQueue.queue,
msg => msg && resSubject$.next({ msg, tag: consumer.consumerTag }),
{ noAck: true },
);
+ setTimeout(() => {
+ channel
+ .cancel(consumer.consumerTag)
+ .catch(error => resSubject$.error(error));
+ }, timeout);
+
replyToSubject.next(replyQueue.queue);
});
@@ -139,6 +147,7 @@ class AmqpStrategy implements TransportLayer {
return ({
host: this.options.host,
channel: this.options.queue,
+ timeout: this.options.timeout ?? DEFAULT_TIMEOUT,
});
}

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.interface.ts
packages/messaging/src/transport/strategies/redis.strategy.interface.ts
@@ -10,6 +10,7 @@ export interface RedisStrategyOptions {
channel: string;
port?: number;
password?: string;
+ timeout?: number;
}

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -2,7 +2,7 @@ import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { Subject, fromEvent, merge } from 'rxjs';
import { map, mapTo, take } from 'rxjs/operators';
import { RedisClient, ClientOpts } from 'redis';
-import { TransportLayer, TransportLayerConnection, TransportMessage, Transport } from '../transport.interface';
+import { TransportLayer, TransportLayerConnection, TransportMessage, Transport, DEFAULT_TIMEOUT } from '../transport.interface';
import { throwUnsupportedError } from '../transport.error';
import { RedisStrategyOptions, RedisConnectionStatus } from './redis.strategy.interface';
import * as RedisHelper from './redis.strategy.helper';
@@ -117,6 +117,7 @@ class RedisStrategy implements TransportLayer {
return ({
host: this.options.host,
channel: this.options.channel,
+ timeout: this.options.timeout ?? DEFAULT_TIMEOUT,
});
}

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -4,6 +4,8 @@ import { RedisStrategy } from './strategies/redis.strategy.interface';
import { AmqpStrategy } from './strategies/amqp.strategy.interface';
import { TcpStrategy } from './strategies/tcp.strategy.interface';
+export const DEFAULT_TIMEOUT = 120 * 1000;
+
export enum Transport {
TCP,
NATS,
@@ -28,6 +30,7 @@ export interface TransportLayer {
export interface TransportLayerConfig {
host: string;
channel: string;
+ timeout: number;
}
export interface TransportLayerConnection {

feat(messaging): timeout RPC events and cleanup AMQP resources