test(messaging): eventBus
MODIFY
packages/core/src/+internal/utils/any.util.ts
packages/core/src/+internal/utils/any.util.ts
@@ -15,3 +15,8 @@ export const getPortEnv = () => pipe(
O.map(Number),
O.toUndefined,
);
+
+export const wait = (seconds = 1) =>
+ new Promise(res => {
+ setTimeout(() => res(), seconds * 1000);
+ });

ADD
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
+import { Event, matchEvent, use, combineEffects } from '@marblejs/core';
+import { wait } from '@marblejs/core/dist/+internal/utils';
+import { eventValidator$, t } from '@marblejs/middleware-io';
+import { delay, map, tap } from 'rxjs/operators';
+import { MsgEffect } from '../effects/messaging.effects.interface';
+import { runEventBus, runEventBusClient } from '../util/messaging.test.util';
+import { reply } from '../effects/messaging.effects.helper';
+
+describe('#eventBus', () => {
+ test('handles RPC event', async () => {
+ const rpc$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_TEST'),
+ use(eventValidator$(t.number)),
+ delay(10),
+ map(event => reply(event)({ type: 'RPC_TEST_RESULT', payload: event.payload + 1 })),
+ );
+
+ const eventBus = await runEventBus(rpc$);
+ const eventBusClient = await runEventBusClient();
+ const event: Event = { type: 'RPC_TEST', payload: 1 };
+
+ const result = await eventBusClient.send(event).toPromise();
+
+ expect(result).toEqual({ type: 'RPC_TEST_RESULT', payload: 2 });
+
+ await eventBus.close();
+ await eventBusClient.close();
+ });
+
+ test('handles parallel RPC events', async () => {
+ const rpc1$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_1_TEST'),
+ use(eventValidator$(t.number)),
+ delay(10),
+ map(event => reply(event)({ type: 'RPC_1_TEST_RESULT', payload: event.payload + 1 })),
+ );
+
+ const rpc2$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_2_TEST'),
+ use(eventValidator$(t.number)),
+ delay(10),
+ map(event => reply(event)({ type: 'RPC_2_TEST_RESULT', payload: event.payload + 1 })),
+ );
+
+ const eventBus = await runEventBus(combineEffects(rpc1$, rpc2$));
+ const eventBusClient = await runEventBusClient();
+
+ const event1: Event = { type: 'RPC_1_TEST', payload: 1 };
+ const event2: Event = { type: 'RPC_2_TEST', payload: 2 };
+ const event3: Event = { type: 'RPC_1_TEST', payload: 3 };
+ const event4: Event = { type: 'RPC_2_TEST', payload: 4 };
+
+ const result = await Promise.all([
+ eventBusClient.send(event1).toPromise(),
+ eventBusClient.send(event2).toPromise(),
+ eventBusClient.send(event3).toPromise(),
+ eventBusClient.send(event4).toPromise(),
+ ]);
+
+ expect(result).toEqual([
+ { type: 'RPC_1_TEST_RESULT', payload: 2 },
+ { type: 'RPC_2_TEST_RESULT', payload: 3 },
+ { type: 'RPC_1_TEST_RESULT', payload: 4 },
+ { type: 'RPC_2_TEST_RESULT', payload: 5 },
+ ]);
+
+ await eventBus.close();
+ await eventBusClient.close();
+ });
+
+ test('handles published event', async done => {
+ const foo$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('TEST'),
+ tap(async event => {
+ expect(event.type).toEqual('TEST');
+ expect(event.payload).toEqual(1);
+
+ await wait();
+ await eventBus.close();
+ await eventBusClient.close();
+ done();
+ }),
+ );
+
+ const eventBus = await runEventBus(foo$);
+ const eventBusClient = await runEventBusClient();
+ const event: Event = { type: 'TEST', payload: 1 };
+
+ await eventBusClient.emit(event);
+ });
+});

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
@@ -10,7 +10,7 @@ import { messagingListener } from '../server/messaging.server.listener';
import { MessagingClient } from '../client/messaging.client.interface';
import { messagingClient } from '../client/messaging.client';
-interface EventBusConfig {
+export interface EventBusConfig {
listener: ReturnType<typeof messagingListener>;
}

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -5,7 +5,7 @@ import { map, tap, delay, bufferCount, mergeMap } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
import { MsgEffect, MsgErrorEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
import { AmqpStrategyOptions } from '../../transport/strategies/amqp.strategy.interface';
-import { runClient, runServer, createMessage } from '../../util/messaging.test.util';
+import { runMicroserviceClient, runMicroservice, createMessage } from '../../util/messaging.test.util';
const createOptions = (config: { expectAck?: boolean; queue?: string } = {}): AmqpStrategyOptions => ({
host: 'amqp://localhost:5672',
@@ -25,8 +25,8 @@ describe('messagingServer::AMQP', () => {
);
const options = createOptions({ expectAck: false });
- const client = await runClient(Transport.AMQP, options);
- const server = await runServer(Transport.AMQP, options)(rpc$);
+ const client = await runMicroserviceClient(Transport.AMQP, options);
+ const microservice = await runMicroservice(Transport.AMQP, options)(rpc$);
const result = await Promise.all([
client.sendMessage(options.queue, createMessage({ type: 'RPC_TEST', payload: 1 })),
@@ -39,7 +39,7 @@ describe('messagingServer::AMQP', () => {
expect(parsedResult0).toEqual({ type: 'RPC_TEST_RESULT', payload: 2 });
expect(parsedResult1).toEqual({ type: 'RPC_TEST_RESULT', payload: 11 });
- await server.close();
+ await microservice.close();
});
test('handles published event', async done => {
@@ -47,7 +47,7 @@ describe('messagingServer::AMQP', () => {
eventSubject.subscribe(event => {
expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
- setTimeout(() => server.close().then(done), 1000);
+ setTimeout(() => microservice.close().then(done), 1000);
});
const event$: MsgEffect = event$ =>
@@ -60,8 +60,8 @@ describe('messagingServer::AMQP', () => {
);
const options = createOptions({ expectAck: false });
- const client = await runClient(Transport.AMQP, options);
- const server = await runServer(Transport.AMQP, options)(event$);
+ const client = await runMicroserviceClient(Transport.AMQP, options);
+ const microservice = await runMicroservice(Transport.AMQP, options)(event$);
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
await client.emitMessage(options.queue, message);
@@ -72,7 +72,7 @@ describe('messagingServer::AMQP', () => {
eventSubject.subscribe(event => {
expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
- setTimeout(() => server.close().then(done), 1000);
+ setTimeout(() => microservice.close().then(done), 1000);
});
const event$: MsgEffect = (event$, { client }) =>
@@ -87,8 +87,8 @@ describe('messagingServer::AMQP', () => {
);
const options = createOptions({ expectAck: true, queue: 'test_ack_queue_server' });
- const client = await runClient(Transport.AMQP, options);
- const server = await runServer(Transport.AMQP, options)(event$);
+ const client = await runMicroserviceClient(Transport.AMQP, options);
+ const microservice = await runMicroservice(Transport.AMQP, options)(event$);
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
await client.emitMessage(options.queue, message);
@@ -107,7 +107,7 @@ describe('messagingServer::AMQP', () => {
expect(data[0][1]).toEqual(error);
expect(data[1][0]).toEqual(event2);
expect(data[1][1]).toBeUndefined();
- setTimeout(() => server.close().then(done), 1000);
+ setTimeout(() => microservice.close().then(done), 1000);
});
const event1$: MsgEffect = event$ =>
@@ -134,8 +134,8 @@ describe('messagingServer::AMQP', () => {
);
const options = createOptions({ expectAck: false });
- const client = await runClient(Transport.AMQP, options);
- const server = await runServer(Transport.AMQP, options)(combineEffects(event1$, event2$), output$, error$);
+ const client = await runMicroserviceClient(Transport.AMQP, options);
+ const microservice = await runMicroservice(Transport.AMQP, options)(combineEffects(event1$, event2$), output$, error$);
await client.emitMessage(options.queue, createMessage(event1));
await client.emitMessage(options.queue, createMessage(event2));

MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
import { matchEvent, use } from '@marblejs/core';
+import { wait } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
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, wait } from '../../util/messaging.test.util';
+import { runMicroserviceClient, runMicroservice, createMessage } from '../../util/messaging.test.util';
const createOptions = (config: { channel?: string } = {}): RedisStrategyOptions => ({
host: 'redis://127.0.0.1:6379',
@@ -18,8 +19,8 @@ describe('messagingServer::Redis', () => {
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)();
+ const client = await runMicroserviceClient(Transport.REDIS, options);
+ const server = await runMicroservice(Transport.REDIS, options)();
await client.close();
await server.close();
@@ -34,8 +35,8 @@ describe('messagingServer::Redis', () => {
);
const options = createOptions();
- const client = await runClient(Transport.REDIS, options);
- const server = await runServer(Transport.REDIS, options)(rpc$);
+ const client = await runMicroserviceClient(Transport.REDIS, options);
+ const microservice = await runMicroservice(Transport.REDIS, options)(rpc$);
const message = createMessage({ type: 'RPC_TEST', payload: 1 });
const result = await client.sendMessage(options.channel, message);
@@ -43,7 +44,7 @@ describe('messagingServer::Redis', () => {
expect(parsedResult).toEqual({ type: 'RPC_TEST_RESULT', payload: 2 });
- await server.close();
+ await microservice.close();
await client.close();
});
@@ -57,15 +58,15 @@ describe('messagingServer::Redis', () => {
tap(async event => {
expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
await wait();
- await server.close();
+ await microservice.close();
await client.close();
done();
}),
);
const options = createOptions();
- const client = await runClient(Transport.REDIS, options);
- const server = await runServer(Transport.REDIS, options)(event$);
+ const client = await runMicroserviceClient(Transport.REDIS, options);
+ const microservice = await runMicroservice(Transport.REDIS, options)(event$);
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
await client.emitMessage(options.channel, message);
@@ -90,15 +91,15 @@ describe('messagingServer::Redis', () => {
expect(event.type).toEqual('UNHANDLED_ERROR');
expect(event.error).toEqual(expectedEventError);
await wait();
- await server.close();
+ await microservice.close();
await client.close();
done();
}),
);
const options = createOptions();
- const client = await runClient(Transport.REDIS, options);
- const server = await runServer(Transport.REDIS, options)(event$, undefined, error$);
+ const client = await runMicroserviceClient(Transport.REDIS, options);
+ const microservice = await runMicroservice(Transport.REDIS, options)(event$, undefined, error$);
const message = createMessage({ type: 'EVENT_TEST' });
const emitResult = await client.emitMessage(options.channel, message);

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
import * as uuid from 'uuid/v4';
+import { createContext, register, bindTo, LoggerToken, mockLogger } from '@marblejs/core';
+import { flow } from 'fp-ts/lib/function';
import { createMicroservice } from '../server/messaging.server';
import { TransportMessage, Transport } from '../transport/transport.interface';
import { MsgEffect, MsgOutputEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
import { messagingListener } from '../server/messaging.server.listener';
import { provideTransportLayer } from '../transport/transport.provider';
+import { eventBus, eventBusClient } from '../eventbus/messaging.eventBus.reader';
-export const runServer = (transport: any, options: any) => (effect$?: MsgEffect, output$?: MsgOutputEffect, error$?: MsgErrorEffect) =>
- createMicroservice({
- options,
- transport,
- listener: messagingListener(effect$
- ? { effects: [effect$], output$, error$ }
- : undefined),
- }).then(app => app());
+// event bus
+export const runEventBus =
+ async (effect$?: MsgEffect, output$?: MsgOutputEffect, error$?: MsgErrorEffect) => {
+ const boundLogger = bindTo(LoggerToken)(mockLogger);
+ const context = flow(register(boundLogger))(createContext());
+ const listener = messagingListener({ effects: effect$ ? [effect$] : [], output$, error$ });
+ return eventBus({ listener })(context);
+ };
-export const runClient = (transport: Transport, transportOptions: any) =>
+export const runEventBusClient = () => eventBusClient(createContext());
+
+// microservice
+export const runMicroservice = (transport: any, options: any) =>
+ async (effect$?: MsgEffect, output$?: MsgOutputEffect, error$?: MsgErrorEffect) => {
+ const listener = messagingListener({ effects: effect$ ? [effect$] : [], output$, error$ });
+ return (await createMicroservice({ options, transport, listener }))();
+ };
+
+export const runMicroserviceClient = (transport: Transport, transportOptions: any) =>
provideTransportLayer(transport, transportOptions).connect({ isConsumer: false });
+// other
export const createMessage = (data: any): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
correlationId: uuid(),
});
-export const wait = (seconds = 1) =>
- new Promise(res => {
- setTimeout(() => res(), seconds * 1000);
- });
-

test(messaging): eventBus