fix(messaging): hanging connection closing
MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
@@ -4,7 +4,7 @@ import { map, tap } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
import { MsgEffect, MsgErrorEffect } from '../../effects/messaging.effects.interface';
import { RedisStrategyOptions } from '../../transport/strategies/redis.strategy.interface';
-import { runClient, runServer, createMessage, close } from '../../util/messaging.test.util';
+import { runClient, runServer, createMessage, wait } from '../../util/messaging.test.util';
const createOptions = (config: { channel?: string } = {}): RedisStrategyOptions => ({
host: 'redis://127.0.0.1:6379',
@@ -16,6 +16,15 @@ describe('messagingServer::Redis', () => {
jest.spyOn(console, 'error').mockImplementation(() => jest.fn());
});
+ test('starts a server and closes connection immediately', async () => {
+ const options = createOptions();
+ const client = await runClient(Transport.REDIS, options);
+ const server = await runServer(Transport.REDIS, options)();
+
+ await client.close();
+ await server.close();
+ });
+
test('handles RPC event', async () => {
const rpc$: MsgEffect = event$ =>
event$.pipe(
@@ -36,6 +45,7 @@ describe('messagingServer::Redis', () => {
expect(parsedResult).toEqual({ type: 'RPC_TEST_RESULT', payload: 2 });
await server.close();
+ await client.close();
});
test('handles published event', async done => {
@@ -45,9 +55,12 @@ describe('messagingServer::Redis', () => {
use(eventValidator$(t.number)),
map(event => event.payload),
map(payload => ({ type: 'EVENT_TEST_RESPONSE', payload: payload + 1 })),
- tap(event => {
+ tap(async event => {
expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
- close(server)(done);
+ await wait();
+ await server.close();
+ await client.close();
+ done();
}),
);
@@ -69,10 +82,13 @@ describe('messagingServer::Redis', () => {
const error$: MsgErrorEffect = event$ =>
event$.pipe(
- tap(({ event, error }) => {
+ tap(async ({ event, error }) => {
expect(event).toBeUndefined();
expect(error).toEqual(error);
- close(server)(done);
+ await wait();
+ await server.close();
+ await client.close();
+ done();
}),
map(() => createEvent('ERROR')),
);

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.helper.ts
packages/messaging/src/transport/strategies/redis.strategy.helper.ts
import * as redis from 'redis';
import { TransportMessage } from '../transport.interface';
-export const connectRedisClient = (options?: redis.ClientOpts | undefined): Promise<redis.RedisClient> =>
+export const connectClient = (options?: redis.ClientOpts | undefined): Promise<redis.RedisClient> =>
new Promise((resolve, reject) => {
const client = redis.createClient(options);
- client.on('connect', error => error ? reject() : resolve(client));
+ client.once('connect', error => error ? reject() : resolve(client));
});
-export const quitRedisClient = (client: redis.RedisClient): Promise<undefined> =>
+export const quitClient = (client: redis.RedisClient): Promise<undefined> =>
new Promise((resolve, reject) => {
client.quit(error => error ? reject() : resolve());
});
-export const subscribeRedisChannel = (client: redis.RedisClient) => (channel: string): Promise<undefined> =>
+export const setExpirationForChannel = (client: redis.RedisClient) => (channel: string) => (seconds: number): Promise<undefined> =>
+ new Promise((resolve, reject) => {
+ client.expire(channel, seconds, error => error ? reject() : resolve());
+ });
+
+export const subscribeChannel = (client: redis.RedisClient) => (channel: string): Promise<undefined> =>
new Promise((resolve, reject) => {
client.subscribe(channel, error => error ? reject() : resolve());
});
-export const unsubscribeRedisChannel = (client: redis.RedisClient) => (channel: string): Promise<undefined> =>
+export const unsubscribeChannel = (client: redis.RedisClient) => (channel: string): Promise<undefined> =>
new Promise((resolve, reject) => {
client.unsubscribe(channel, error => error ? reject() : resolve());
});
+export const publishMessage = (client: redis.RedisClient) => (channel: string) => (message: string): Promise<boolean> =>
+ new Promise((res, rej) => {
+ client.publish(channel, message, err => err ? rej(false) : res(true));
+ });
+
export const encodeMessage = (uuid: string | undefined) => (message: Buffer): string =>
`${uuid}::${message.toString()}`;

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -4,8 +4,8 @@ import { map, mapTo, take } from 'rxjs/operators';
import { RedisClient, ClientOpts } from 'redis';
import { TransportLayer, TransportLayerConnection, TransportMessage } from '../transport.interface';
import { RedisStrategyOptions, RedisConnectionStatus } from './redis.strategy.interface';
-import { quitRedisClient, subscribeRedisChannel, encodeMessage, decodeMessage, unsubscribeRedisChannel, connectRedisClient } from './redis.strategy.helper';
import { throwUnsupportedError } from '../transport.error';
+import * as RedisHelper from './redis.strategy.helper';
class RedisStrategyConnection implements TransportLayerConnection {
private msgSubject$ = new Subject<{ content: string }>();
@@ -46,21 +46,21 @@ class RedisStrategyConnection implements TransportLayerConnection {
return merge(
fromEvent<Error>(this.publisher, 'error'),
fromEvent<Error>(this.subscriber, 'error'),
+ fromEvent<Error>(this.rpcSubscriber, 'error'),
);
}
get message$() {
return this.msgSubject$.asObservable().pipe(
- map(msg => decodeMessage(msg.content)),
+ map(msg => RedisHelper.decodeMessage(msg.content)),
);
}
- emitMessage = async (channel: string, message: TransportMessage<Buffer>) =>
- new Promise<boolean>((res, rej) => {
+ emitMessage = async (channel: string, message: TransportMessage<Buffer>) => {
const replyChannel = message.correlationId;
- const encodedMessage = encodeMessage(replyChannel)(message.data);
- this.publisher.publish(channel, encodedMessage, err => err ? rej(false) : res(true));
- });
+ const encodedMessage = RedisHelper.encodeMessage(replyChannel)(message.data);
+ return RedisHelper.publishMessage(this.publisher)(channel)(encodedMessage);
+ }
sendMessage = async (channel: string, message: TransportMessage<Buffer>) => {
const rpcSubject$ = new Subject<{ content: string }>();
@@ -69,14 +69,13 @@ class RedisStrategyConnection implements TransportLayerConnection {
message.correlationId = correlationId;
- this.rpcSubscriber.expire(replyChannel, 1);
+ await RedisHelper.setExpirationForChannel(this.rpcSubscriber)(replyChannel)(1);
+ await RedisHelper.subscribeChannel(this.rpcSubscriber)(replyChannel);
- await subscribeRedisChannel(this.rpcSubscriber)(replyChannel);
-
- this.rpcSubscriber.on('message', async (ch, msg) => {
+ this.rpcSubscriber.once('message', async (ch, msg) => {
if (ch === replyChannel) {
rpcSubject$.next({ content: msg });
- await unsubscribeRedisChannel(this.rpcSubscriber)(replyChannel);
+ await RedisHelper.unsubscribeChannel(this.rpcSubscriber)(replyChannel);
}
});
@@ -84,14 +83,17 @@ class RedisStrategyConnection implements TransportLayerConnection {
return rpcSubject$.asObservable().pipe(
take(1),
- map(msg => decodeMessage(msg.content)),
+ map(msg => RedisHelper.decodeMessage(msg.content)),
).toPromise();
}
close = async () => {
+ this.subscriber.removeAllListeners('message');
+
await Promise.all([
- quitRedisClient(this.publisher),
- quitRedisClient(this.subscriber),
+ RedisHelper.quitClient(this.publisher),
+ RedisHelper.quitClient(this.subscriber),
+ RedisHelper.quitClient(this.rpcSubscriber),
]);
this.closeSubject$.next();
@@ -125,15 +127,15 @@ class RedisStrategy implements TransportLayer {
await import('redis');
- const publisher = await connectRedisClient(this.clientOpts);
- const subscriber = await connectRedisClient(this.clientOpts);
- const rpcSubscriber = await connectRedisClient(this.clientOpts);
+ const publisher = await RedisHelper.connectClient(this.clientOpts);
+ const subscriber = await RedisHelper.connectClient(this.clientOpts);
+ const rpcSubscriber = await RedisHelper.connectClient(this.clientOpts);
- publisher.expire(channel, 1);
+ await RedisHelper.setExpirationForChannel(publisher)(channel)(1);
if (data.isConsumer) {
- subscriber.expire(channel, 1);
- await subscribeRedisChannel(subscriber)(channel);
+ await RedisHelper.setExpirationForChannel(subscriber)(channel)(1);
+ await RedisHelper.subscribeChannel(subscriber)(channel);
}
return new RedisStrategyConnection({ isConsumer, channel }, publisher, subscriber, rpcSubscriber);

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
import * as uuid from 'uuid/v4';
import { createMicroservice } from '../server/messaging.server';
-import { TransportMessage, Transport, TransportLayerConnection } from '../transport/transport.interface';
+import { TransportMessage, Transport } from '../transport/transport.interface';
import { MsgEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
import { messagingListener } from '../server/messaging.server.listener';
import { provideTransportLayer } from '../transport/transport.provider';
@@ -20,5 +20,8 @@ export const createMessage = (data: any): TransportMessage<Buffer> => ({
correlationId: uuid(),
});
-export const close = (conn: TransportLayerConnection) => (done: () => void) =>
- setTimeout(() => conn.close().then(done), 1000);
+export const wait = (seconds = 1000) =>
+ new Promise(res => {
+ setTimeout(() => res(), seconds);
+ });
+

fix(messaging): hanging connection closing