fix(messaging): lazyload `redis` module
MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -34,10 +34,11 @@ export interface MessagingListenerConfig {
const defaultOutput$: MsgOutputEffect = msg$ => msg$.pipe(map(m => m.event));
const defaultError$: MsgErrorEffect = msg$ => msg$;
+const defaultEffect$: MsgEffect = msg$ => msg$;
export const messagingListener = (config: MessagingListenerConfig = {}) => {
const {
- effects = [],
+ effects = [defaultEffect$],
middlewares = [],
output$ = defaultOutput$,
error$ = defaultError$,

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 { RedisClient, ClientOpts } from 'redis';
import { TransportMessage } from '../transport.interface';
-export const connectClient = (options?: redis.ClientOpts | undefined): Promise<redis.RedisClient> =>
- new Promise((resolve, reject) => {
+export const connectClient = async (options?: ClientOpts | undefined): Promise<RedisClient> => {
+ const redis = await import('redis');
+
+ return new Promise((resolve, reject) => {
const client = redis.createClient(options);
client.once('connect', error => error ? reject() : resolve(client));
});
+}
-export const quitClient = (client: redis.RedisClient): Promise<undefined> =>
+export const quitClient = (client: RedisClient): Promise<undefined> =>
new Promise((resolve, reject) => {
client.quit(error => error ? reject() : resolve());
});
-export const setExpirationForChannel = (client: redis.RedisClient) => (channel: string) => (seconds: number): Promise<undefined> =>
+export const setExpirationForChannel = (client: 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> =>
+export const subscribeChannel = (client: RedisClient) => (channel: string): Promise<undefined> =>
new Promise((resolve, reject) => {
client.subscribe(channel, error => error ? reject() : resolve());
});
-export const unsubscribeChannel = (client: redis.RedisClient) => (channel: string): Promise<undefined> =>
+export const unsubscribeChannel = (client: 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> =>
+export const publishMessage = (client: RedisClient) => (channel: string) => (message: string): Promise<boolean> =>
new Promise((res, rej) => {
client.publish(channel, message, err => err ? rej(false) : res(true));
});
@@ -40,6 +43,6 @@ export const decodeMessage = (message: string): TransportMessage<Buffer> => {
return {
replyTo: splittedMsg[0],
data: Buffer.from(splittedMsg[1]),
- }
+ };
};

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -125,8 +125,6 @@ class RedisStrategy implements TransportLayer {
const { isConsumer } = data;
const { channel } = this.options;
- await import('redis');
-
const publisher = await RedisHelper.connectClient(this.clientOpts);
const subscriber = await RedisHelper.connectClient(this.clientOpts);
const rpcSubscriber = await RedisHelper.connectClient(this.clientOpts);

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
@@ -20,8 +20,8 @@ export const createMessage = (data: any): TransportMessage<Buffer> => ({
correlationId: uuid(),
});
-export const wait = (seconds = 1000) =>
+export const wait = (seconds = 1) =>
new Promise(res => {
- setTimeout(() => res(), seconds);
+ setTimeout(() => res(), seconds * 1000);
});

fix(messaging): lazyload `redis` module