fix(messaging): redis RPC communication
MODIFY
packages/@integration/package.json
packages/@integration/package.json
"start:cqrs": "PORT=1337 ts-node --project tsconfig.json src/cqrs/index.ts",
"start:websockets": "ts-node --project tsconfig.json src/websockets/http.server.ts",
"start:messaging:server": "PORT=1337 ts-node --project tsconfig.json src/messaging/server.ts",
- "start:messaging:client": "ts-node --project tsconfig.json src/messaging/client.ts",
+ "start:messaging:client": "PORT=1338 ts-node --project tsconfig.json src/messaging/client.ts",
"watch:dev": "nodemon --watch src --ext ts --ignore '*.spec.ts' --exec yarn start",
"test": "jest --config ../../jest.config.js",
"clean": "rimraf dist"

MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
import { logger$ } from '@marblejs/middleware-logger';
import { requestValidator$, t } from '@marblejs/middleware-io';
import { messagingClient, MessagingClient, Transport } from '@marblejs/messaging';
-import { r, createServer, createContextToken, httpListener, use, useContext, bindEagerlyTo } from '@marblejs/core';
-import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
+import { r, createServer, createContextToken, httpListener, use, useContext, bindEagerlyTo, combineRoutes, ContextToken } from '@marblejs/core';
+import { isTestEnv, getPortEnv } from '@marblejs/core/dist/+internal/utils';
import { IO } from 'fp-ts/lib/IO';
import { forkJoin } from 'rxjs';
-import { map, mergeMap, mapTo } from 'rxjs/operators';
+import { map, mergeMap } from 'rxjs/operators';
-const port = process.env.PORT
- ? Number(process.env.PORT)
- : undefined;
+const AmqpClientToken = createContextToken<MessagingClient>('AmqpMessagingClient');
+const RedisClientToken = createContextToken<MessagingClient>('RedisMessagingClient');
-const ClientToken = createContextToken<MessagingClient>('MessagingClient');
-
-const client = messagingClient({
+const amqpClient = messagingClient({
transport: Transport.AMQP,
options: {
host: 'amqp://localhost:5672',
queue: 'test_queue',
queueOptions: { durable: false },
- timeout: isTestEnv()
- ? 500
- : 30 * 1000,
+ timeout: isTestEnv() ? 500 : 30 * 1000,
+ },
+});
+
+const redisClient = messagingClient({
+ transport: Transport.REDIS,
+ options: {
+ host: 'redis://127.0.0.1:6379',
+ channel: 'test_channel',
+ timeout: isTestEnv() ? 500 : 30 * 1000,
},
});
@@ -31,50 +35,11 @@ const rootValiadtor$ = requestValidator$({
}),
});
-const buffer$ = r.pipe(
- r.matchPath('/buffer'),
- r.matchType('GET'),
- r.useEffect((req$, { ask }) => {
- const client = useContext(ClientToken)(ask);
-
- return req$.pipe(
- mergeMap(() => client.emit({ type: 'BUFFER' })),
- mapTo(({ body: 'OK' })),
- );
- }),
-);
-
-const timeout$ = r.pipe(
- r.matchPath('/timeout'),
- r.matchType('GET'),
- r.useEffect((req$, { ask }) => {
- const client = useContext(ClientToken)(ask);
-
- return req$.pipe(
- mergeMap(() => client.send({ type: 'TIMEOUT' })),
- mapTo(({ body: 'OK' })),
- );
- }),
-);
-
-const error$ = r.pipe(
- r.matchPath('/error'),
- r.matchType('GET'),
- r.useEffect((req$, { ask }) => {
- const client = useContext(ClientToken)(ask);
-
- return req$.pipe(
- mergeMap(() => client.send({ type: 'ERROR' })),
- mapTo(({ body: 'OK' })),
- );
- }),
-);
-
-const fib$ = r.pipe(
+const fib$ = (clientToken: ContextToken<MessagingClient>) => r.pipe(
r.matchPath('/fib/:number'),
r.matchType('GET'),
r.useEffect((req$, ctx) => {
- const client = useContext(ClientToken)(ctx.ask);
+ const client = useContext(clientToken)(ctx.ask);
return req$.pipe(
use(rootValiadtor$),
@@ -91,14 +56,23 @@ const fib$ = r.pipe(
}),
);
+const amqp$ = combineRoutes('/amqp', [
+ fib$(AmqpClientToken),
+]);
+
+const redis$ = combineRoutes('/redis', [
+ fib$(RedisClientToken),
+]);
+
export const server = createServer({
- port,
+ port: getPortEnv(),
listener: httpListener({
middlewares: [logger$()],
- effects: [buffer$, timeout$, error$, fib$],
+ effects: [amqp$, redis$],
}),
dependencies: [
- bindEagerlyTo(ClientToken)(client),
+ bindEagerlyTo(AmqpClientToken)(amqpClient),
+ bindEagerlyTo(RedisClientToken)(redisClient),
],
});

MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
import { of } from 'rxjs';
-import { map, tap, mergeMap, bufferCount, mapTo, catchError, delay } from 'rxjs/operators';
-import { matchEvent, use } from '@marblejs/core';
+import { map, tap, catchError } from 'rxjs/operators';
+import { matchEvent, use, act } from '@marblejs/core';
+import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { createMicroservice, messagingListener, Transport, MsgEffect, reply } from '@marblejs/messaging';
-import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
import { IO } from 'fp-ts/lib/IO';
const fib = (n: number): number =>
@@ -12,8 +12,8 @@ const fib = (n: number): number =>
const fibonacci$: MsgEffect = event$ =>
event$.pipe(
matchEvent('FIB'),
+ act(event => of(event).pipe(
use(eventValidator$(t.number)),
- mergeMap(event => of(event).pipe(
tap(event => { if (event.payload >= 45) throw new Error('Too high number!') }),
map(event => fib(event.payload)),
map(payload => reply(event)({ type: 'FIB_RESULT', payload })),
@@ -21,44 +21,35 @@ const fibonacci$: MsgEffect = event$ =>
)),
);
-const buffer$: MsgEffect = event$ =>
- event$.pipe(
- matchEvent('BUFFER'),
- bufferCount(2),
- mapTo({ type: 'BUFFER_RESULT' }),
- );
-
-const timeout$: MsgEffect = event$ =>
- event$.pipe(
- matchEvent('TIMEOUT'),
- mergeMap(event => of(event).pipe(
- delay(180 * 1000),
- mapTo(reply(event)({ type: 'TIMEOUT_RESULT' })),
- )),
- );
-
-const error$: MsgEffect = event$ =>
- event$.pipe(
- matchEvent('ERROR'),
- tap(() => { throw new Error('test_error'); }),
- );
-
-export const microservice = createMicroservice({
+export const amqpMicroservice = createMicroservice({
transport: Transport.AMQP,
options: {
host: 'amqp://localhost:5672',
queue: 'test_queue',
queueOptions: { durable: false },
- timeout: isTestEnv()
- ? 500
- : 30 * 1000
+ timeout: isTestEnv() ? 500 : 30 * 1000
+ },
+ listener: messagingListener({
+ effects: [fibonacci$],
+ }),
+});
+
+export const redisMicroservice = createMicroservice({
+ transport: Transport.REDIS,
+ options: {
+ host: 'redis://127.0.0.1:6379',
+ channel: 'test_channel',
+ timeout: isTestEnv() ? 500 : 30 * 1000
},
listener: messagingListener({
- effects: [fibonacci$, buffer$, timeout$, error$],
+ effects: [fibonacci$],
}),
});
const main: IO<void> = async () =>
- !isTestEnv() && await (await microservice)();
+ !isTestEnv() && await Promise.all([
+ (await amqpMicroservice)(),
+ (await redisMicroservice)(),
+ ]);
main();

MODIFY
packages/@integration/test/messaging.integration.spec.ts
packages/@integration/test/messaging.integration.spec.ts
@@ -4,27 +4,30 @@ import * as request from 'supertest';
import { createHttpServerTestBed } from '@marblejs/core/dist/+internal/testing';
import { createMicroserviceTestBed } from '@marblejs/messaging/dist/+internal/testing';
import { server } from '../src/messaging/client';
-import { microservice } from '../src/messaging/server';
+import { amqpMicroservice, redisMicroservice } from '../src/messaging/server';
describe('messaging integration', () => {
const httpTestBed = createHttpServerTestBed(server);
- const microserviceTestBed = createMicroserviceTestBed(microservice);
+
+ createMicroserviceTestBed(amqpMicroservice);
+ createMicroserviceTestBed(redisMicroservice);
beforeAll(async () => {
jest.spyOn(console, 'log').mockImplementation();
jest.spyOn(console, 'info').mockImplementation();
});
- test('GET /fib returns 10, 11, 12, 13, 14 th fibonacci number', async () =>
+ test.each([
+ 'redis',
+ 'amqp',
+ ])('GET /%s/fib returns 10, 11, 12, 13, 14 th fibonacci number', async type =>
request(httpTestBed.getInstance())
- .get('/fib/10')
+ .get(`/${type}/fib/10`)
.expect(200, [
{ type: 'FIB_RESULT', payload: 55 },
{ type: 'FIB_RESULT', payload: 89 },
{ type: 'FIB_RESULT', payload: 144 },
{ type: 'FIB_RESULT', payload: 233 },
{ type: 'FIB_RESULT', payload: 377 },
- ])
- );
+ ]));
});
-

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { Subject, fromEvent, merge } from 'rxjs';
-import { map, mapTo, take } from 'rxjs/operators';
+import { map, mapTo, take, tap, share, filter } from 'rxjs/operators';
import { RedisClient, ClientOpts } from 'redis';
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';
+type RedisIncomingMsg = {
+ content: string;
+ channel: string;
+};
+
class RedisStrategyConnection implements TransportLayerConnection {
- private msgSubject$ = new Subject<{ content: string }>();
+ private producerSubject = new Subject<RedisIncomingMsg>();
+ private consumerSubject = new Subject<RedisIncomingMsg>();
private closeSubject$ = new Subject();
constructor(
@@ -18,7 +24,9 @@ class RedisStrategyConnection implements TransportLayerConnection {
private rpcSubscriber: RedisClient,
) {
if (opts.isConsumer) {
- subscriber.on('message', (_, msg) => this.msgSubject$.next({ content: msg }));
+ subscriber.on('message', (channel, content) => this.consumerSubject.next({ content, channel }));
+ } else {
+ rpcSubscriber.on('message', (channel, content) => this.producerSubject.next({ content, channel }));
}
}
@@ -35,7 +43,9 @@ class RedisStrategyConnection implements TransportLayerConnection {
const end$ = merge(fromEvent(this.subscriber, 'end'), fromEvent(this.publisher, 'end'))
.pipe(mapTo(RedisConnectionStatus.END));
- return merge(ready$, connect$, reconnecting$, end$);
+ return merge(ready$, connect$, reconnecting$, end$).pipe(
+ share(),
+ );
}
get close$() {
@@ -47,14 +57,16 @@ class RedisStrategyConnection implements TransportLayerConnection {
fromEvent<Error>(this.publisher, 'error'),
fromEvent<Error>(this.subscriber, 'error'),
fromEvent<Error>(this.rpcSubscriber, 'error'),
+ ).pipe(
+ share(),
);
}
get message$() {
- return this.msgSubject$.asObservable().pipe(
+ return this.consumerSubject.asObservable().pipe(
map(msg => RedisHelper.decodeMessage(msg.content)),
);
- }
+ }z
emitMessage = async (channel: string, message: TransportMessage<Buffer>) => {
const replyChannel = message.correlationId;
@@ -63,26 +75,18 @@ class RedisStrategyConnection implements TransportLayerConnection {
}
sendMessage = async (channel: string, message: TransportMessage<Buffer>) => {
- const rpcSubject$ = new Subject<{ content: string }>();
const correlationId = createUuid();
const replyChannel = correlationId;
message.correlationId = correlationId;
- await RedisHelper.setExpirationForChannel(this.rpcSubscriber)(replyChannel)(1);
await RedisHelper.subscribeChannel(this.rpcSubscriber)(replyChannel);
-
- this.rpcSubscriber.once('message', async (ch, msg) => {
- if (ch === replyChannel) {
- rpcSubject$.next({ content: msg });
- await RedisHelper.unsubscribeChannel(this.rpcSubscriber)(replyChannel);
- }
- });
-
await this.emitMessage(channel, message);
- return rpcSubject$.asObservable().pipe(
+ return this.producerSubject.asObservable().pipe(
+ filter(msg => msg.channel === replyChannel),
take(1),
+ tap(() => RedisHelper.unsubscribeChannel(this.rpcSubscriber)(replyChannel)),
map(msg => RedisHelper.decodeMessage(msg.content)),
).toPromise();
}

fix(messaging): redis RPC communication