test(messaging): redis client
MODIFY
packages/messaging/src/client/messaging.client.spec.ts
packages/messaging/src/client/messaging.client.spec.ts
import { TimeoutError } from 'rxjs';
import { NamedError } from '@marblejs/core/dist/+internal/utils';
-import { Transport, TransportMessage } from '../transport/transport.interface';
+import { Transport, TransportLayer, TransportLayerConnection } from '../transport/transport.interface';
import { createAmqpStrategy } from '../transport/strategies/amqp.strategy';
-import { createTestContext } from '../util/messaging.test.util';
+import { createTestContext, createMessage } from '../util/messaging.test.util';
+import { createRedisStrategy } from '../transport/strategies/redis.strategy';
import { messagingClient } from './messaging.client';
-import { MessagingClientConfig } from './messaging.client.interface';
+import { MessagingClientConfig, MessagingClient } from './messaging.client.interface';
+
+const runClient = async (config: MessagingClientConfig): Promise<MessagingClient> =>
+ await messagingClient(config)(createTestContext());
+
+const runServer = (transportLayer: TransportLayer): Promise<TransportLayerConnection> =>
+ transportLayer.connect({ isConsumer: true });
describe('#messagingClient', () => {
@@ -19,16 +26,91 @@ describe('#messagingClient', () => {
},
};
- const runClient = async () => await messagingClient(clientOptions)(createTestContext());
- const runServer = () => createAmqpStrategy(clientOptions.options).connect({ isConsumer: true });
- const createMessage = (data: any, correlationId?: string): TransportMessage<Buffer> => ({
- data: Buffer.from(JSON.stringify(data)),
- correlationId: correlationId,
+ const transportLayer = createAmqpStrategy(clientOptions.options);
+
+ test('emits event', async done => {
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
+
+ server.message$.subscribe(async msg => {
+ expect(msg.data.toString()).toEqual(JSON.stringify({ type: 'TEST' }));
+ await server.close();
+ done();
+ });
+
+ await client.emit({ type: 'TEST' });
+ await client.close();
+ });
+
+ test('sends RPC event', async () => {
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
+
+ server.message$.subscribe(async msg => {
+ const inEvent = JSON.parse(msg.data.toString());
+ const outEvent = { ...inEvent, payload: inEvent.payload + 1 };
+ const response = createMessage(outEvent, msg.correlationId);
+
+ await server.emitMessage(msg.replyTo || '', response);
+ await server.close();
+ });
+
+ const result = await client.send({ type: 'TEST', payload: 1 }).toPromise();
+
+ expect(result).toEqual({ type: 'TEST', payload: 2 });
+
+ await client.close();
+ });
+
+ test('timeouts sent RPC event', async () => {
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
+
+ const result = client.send({ type: 'TEST' }).toPromise();
+
+ await expect(result).rejects.toEqual(new TimeoutError());
+
+ await client.close();
+ await server.close();
+ });
+
+ test('throws an error for sent RPC event', async () => {
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
+ const error = { name: 'TestErrorName', message: 'TestErrorMessage' };
+
+ server.message$.subscribe(async msg => {
+ const inEvent = JSON.parse(msg.data.toString());
+ const outEvent = { ...inEvent, error };
+ const response = createMessage(outEvent, msg.correlationId);
+
+ await server.emitMessage(msg.replyTo || '', response);
+ await server.close();
+ });
+
+ const result = client.send(1).toPromise();
+
+ await expect(result).rejects.toEqual(new NamedError(error.name, error.message));
+
+ await client.close();
});
+ });
+
+ describe('Redis', () => {
+ const clientOptions: MessagingClientConfig = {
+ transport: Transport.REDIS,
+ options: {
+ host: 'redis://127.0.0.1:6379',
+ channel: 'test_channel_server',
+ timeout: 1000,
+ },
+ };
+
+ const transportLayer = createRedisStrategy(clientOptions.options);
test('emits event', async done => {
- const client = await runClient();
- const server = await runServer();
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
server.message$.subscribe(async msg => {
expect(msg.data.toString()).toEqual(JSON.stringify({ type: 'TEST' }));
@@ -41,8 +123,8 @@ describe('#messagingClient', () => {
});
test('sends RPC event', async () => {
- const client = await runClient();
- const server = await runServer();
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
server.message$.subscribe(async msg => {
const inEvent = JSON.parse(msg.data.toString());
@@ -61,8 +143,8 @@ describe('#messagingClient', () => {
});
test('timeouts sent RPC event', async () => {
- const client = await runClient();
- const server = await runServer();
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
const result = client.send({ type: 'TEST' }).toPromise();
@@ -73,8 +155,8 @@ describe('#messagingClient', () => {
});
test('throws an error for sent RPC event', async () => {
- const client = await runClient();
- const server = await runServer();
+ const client = await runClient(clientOptions);
+ const server = await runServer(transportLayer);
const error = { name: 'TestErrorName', message: 'TestErrorMessage' };
server.message$.subscribe(async msg => {

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
@@ -29,9 +29,9 @@ export const runMicroserviceClient = (transport: Transport, transportOptions: an
provideTransportLayer(transport, transportOptions).connect({ isConsumer: false });
// other
-export const createMessage = (data: any): TransportMessage<Buffer> => ({
+export const createMessage = (data: any, correlationId?: string): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
- correlationId: uuid(),
+ correlationId: correlationId ?? uuid(),
});
export const createTestContext = () => {

test(messaging): redis client