test(messaging): test cases for amqp server listener
MODIFY
packages/core/src/+internal/utils/spec/string.util.spec.ts
packages/core/src/+internal/utils/spec/string.util.spec.ts
-import { isString, trim, stringify } from '../string.util';
+import { isString, trim, stringify, createUuid } from '../string.util';
test('#isString check is given argument is of string type', () => {
expect(isString('some string value')).toBe(true);
@@ -27,3 +27,7 @@ test('#stringify stringifies given value', () => {
unknownFunction.displayName = 'unknownFunction';
expect(stringify(unknownFunction)).toEqual('unknownFunction');
});
+
+test('#createUuid creates unique identifier', () => {
+ expect(createUuid()).toBeDefined();
+});

MODIFY
packages/messaging/package.json
packages/messaging/package.json
},
"devDependencies": {
"@marblejs/core": "^2.1.0",
+ "@marblejs/middleware-io": "^2.1.0",
"amqplib": "^0.5.3",
"@types/amqplib": "^0.5.11"
},

MODIFY
packages/messaging/src/client/messaging.client.spec.ts
packages/messaging/src/client/messaging.client.spec.ts
@@ -18,6 +18,10 @@ describe('#messagingClient', () => {
const runClient = () => messagingClient(clientOptions).run(createContext());
const runServer = () => createAmqpStrategy(clientOptions.options).connect();
+ const createMessage = (data: any, correlationId?: string): TransportMessage<Buffer> => ({
+ data: Buffer.from(JSON.stringify(data)),
+ correlationId: correlationId,
+ });
test('emits event to consumer', async done => {
const client = runClient();
@@ -40,11 +44,8 @@ describe('#messagingClient', () => {
const server = await runServer();
server.consumeMessage().subscribe(async msg => {
- const count = String(Number(msg.data.toString()) + 1);
- const response: TransportMessage<Buffer> = {
- data: Buffer.from(count),
- correlationId: msg.correlationId,
- };
+ const count = Number(msg.data.toString()) + 1;
+ const response = createMessage(count, msg.correlationId);
await server.emitMessage(msg.replyTo || '', response);
server.ack(msg);

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
import { reader, HttpServerEventStreamToken, matchEvent, ServerEvent } from '@marblejs/core';
import { from, Observable, EMPTY } from 'rxjs';
import { mergeMap, take, map, mapTo, mergeMapTo } from 'rxjs/operators';
-import {
- Transport,
- TransportMessage,
- TransportLayerConnection,
-} from '../transport/transport.interface';
+import { TransportMessage, TransportLayerConnection } from '../transport/transport.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
import { MessagingClient, MessagingClientConfig } from './messaging.client.interface';
@@ -13,9 +9,9 @@ import { createUuid } from '@marblejs/core/dist/+internal/utils';
export const messagingClient = (config: MessagingClientConfig) => {
const {
- transport = Transport.TCP,
+ transport,
+ options,
msgTransformer = jsonTransformer,
- options = {},
} = config;
const emit = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>

ADD
packages/messaging/src/index.spec.ts
+import * as API from './index';
+
+describe('@marblejs/messaging', () => {
+ test('public APIs are defined', () => {
+ expect(API.Transport).toBeDefined();
+ expect(API.TransportLayerToken).toBeDefined();
+ expect(API.createMicroservice).toBeDefined();
+ expect(API.messagingClient).toBeDefined();
+ expect(API.messagingListener).toBeDefined();
+ });
+});

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
-export { messagingListener } from './listener/messaging.listener';
export * from './client/messaging.client';
export * from './client/messaging.client.interface';
export * from './transport/transport.interface';
@@ -6,3 +5,4 @@ export * from './effects/messaging.effects.interface';
export * from './server/messaging.server';
export * from './server/messaging.server.interface';
export * from './server/messaging.server.tokens';
+export { messagingListener } from './server/messaging.server.listener';

MODIFY
packages/messaging/src/server/messaging.server.interface.ts
packages/messaging/src/server/messaging.server.interface.ts
import { BoundDependency } from '@marblejs/core';
-import { messagingListener } from '../listener/messaging.listener';
import { NatsStrategy } from '../transport/strategies/nats.strategy.interface';
import { TcpStrategy } from '../transport/strategies/tcp.strategy.interface';
import { AmqpStrategy } from '../transport/strategies/amqp.strategy.interface';
+import { messagingListener } from './messaging.server.listener';
type ConfigurationBase = {
messagingListener: ReturnType<typeof messagingListener>;

RENAME
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -15,7 +15,7 @@ import {
} from '../transport/transport.interface';
import { jsonTransformer } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
-import { TransportLayerToken } from '../server/messaging.server.tokens';
+import { TransportLayerToken } from './messaging.server.tokens';
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];

ADD
packages/messaging/src/server/messaging.server.spec.ts
+import { matchEvent, use } from '@marblejs/core';
+import { eventValidator$, t } from '@marblejs/middleware-io';
+import { map, tap } from 'rxjs/operators';
+import { createMicroservice } from './messaging.server';
+import { messagingListener } from './messaging.server.listener';
+import { Transport, TransportMessage } from '../transport/transport.interface';
+import { createAmqpStrategy } from '../transport/strategies/amqp.strategy';
+import { MsgEffect } from '../effects/messaging.effects.interface';
+import { Subject } from 'rxjs';
+
+describe('messagingServer', () => {
+
+ describe('AMQP', () => {
+ const options = {
+ host: 'amqp://localhost:5672',
+ queue: 'test_queue_server',
+ queueOptions: { durable: false },
+ };
+
+ const microservice = (effect?: MsgEffect) => createMicroservice({
+ options,
+ transport: Transport.AMQP,
+ messagingListener: messagingListener(effect ? { effects: [effect] } : undefined),
+ });
+
+ const runServer = (effect?: MsgEffect) => microservice(effect).run();
+ const runClient = () => createAmqpStrategy(options).connect();
+ const createMessage = (data: any): TransportMessage<Buffer> => ({
+ data: Buffer.from(JSON.stringify(data)),
+ });
+
+ test('receives RPC response from consumer', async () => {
+ const rpc$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_TEST'),
+ use(eventValidator$(t.number)),
+ map(event => event.payload),
+ map(payload => ({ type: 'RPC_TEST_RESULT', payload: payload + 1 })),
+ );
+
+ const client = await runClient();
+ const server = await runServer(rpc$);
+ const message = createMessage({ type: 'RPC_TEST', payload: 1 });
+
+ const result = await client.sendMessage(options.queue, message);
+ const parsedResult = JSON.parse(result.data.toString());
+
+ expect(parsedResult).toEqual({ type: 'RPC_TEST_RESULT', payload: 2 });
+
+ await server.close();
+ });
+
+ test('emits event to consumer', async done => {
+ const eventSubject = new Subject();
+ const event$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('EVENT_TEST'),
+ use(eventValidator$(t.number)),
+ map(event => event.payload),
+ map(payload => ({ type: 'EVENT_TEST_RESPONSE', payload: payload + 1 })),
+ tap(event => eventSubject.next(event)),
+ );
+
+ const server = await runServer(event$);
+ const client = await runClient();
+ const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
+
+ await client.emitMessage(options.queue, message);
+
+ eventSubject.subscribe(event => {
+ expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
+ setTimeout(() => server.close().then(done), 1000);
+ });
+ });
+ });
+
+});

MODIFY
packages/messaging/src/server/messaging.server.ts
packages/messaging/src/server/messaging.server.ts
import { createContext, registerAll, bindTo } from '@marblejs/core';
import { CreateMicroserviceConfig } from './messaging.server.interface';
import { provideTransportLayer } from '../transport/transport.provider';
-import { Transport } from '../transport/transport.interface';
import { TransportLayerToken } from './messaging.server.tokens';
export const createMicroservice = (config: CreateMicroserviceConfig) => {
const {
- options = {},
+ options,
+ transport,
dependencies = [],
- transport = Transport.AMQP,
messagingListener,
} = config;

MODIFY
packages/messaging/src/transport/strategies/nats.strategy.ts
packages/messaging/src/transport/strategies/nats.strategy.ts
+
import { TransportLayer } from '../transport.interface';
+/* istanbul ignore next */
export const createNatsStrategy = (): TransportLayer => {
// @TODO
return {} as any;

MODIFY
packages/messaging/src/transport/strategies/tcp.strategy.ts
packages/messaging/src/transport/strategies/tcp.strategy.ts
+
import { TransportLayer } from '../transport.interface';
+/* istanbul ignore next */
export const createTcpStrategy = (): TransportLayer => {
// @TODO
return {} as any;

test(messaging): test cases for amqp server listener