feat(messaging): #reply utiltiy function
MODIFY
packages/@integration/src/cqrs/effects/eventbus.effects.ts
packages/@integration/src/cqrs/effects/eventbus.effects.ts
import { map, tap, delay } from 'rxjs/operators';
import { matchEvent, useContext, LoggerToken, LoggerTag } from '@marblejs/core';
-import { MsgEffect, EVENT_BUS_CHANNEL } from '@marblejs/messaging';
+import { MsgEffect, EVENT_BUS_CHANNEL, reply } from '@marblejs/messaging';
import { OfferCommand } from '../domain/Offer.command';
import { OfferEvent } from '../domain/Offer.event';
@@ -13,9 +13,10 @@ export const generateOfferDocument$: MsgEffect = (event$, ctx) => {
matchEvent(OfferCommand.generateOffer),
map(event => event.payload.offerId),
tap(logger({ tag, type: 'generateOfferDocument$', message: 'Generating offer document...'})),
- delay(5 * 1000), // simulate processing
- map(OfferEvent.offerDocumentCreated),
- map(event => ({ ...event, metadata: { replyTo: EVENT_BUS_CHANNEL }})),
+ delay(5 * 1000),
+ map(offerId => reply(EVENT_BUS_CHANNEL)(
+ OfferEvent.offerDocumentCreated(offerId),
+ )),
);
}
@@ -26,7 +27,7 @@ export const saveOfferDocument$: MsgEffect = (event$, ctx) => {
matchEvent(OfferEvent.offerDocumentCreated),
map(event => event.payload.offerId),
tap(logger({ tag, type: 'saveOfferDocument$', message: 'Saving offer document...'})),
- delay(5 * 1000), // simulate processing
+ delay(5 * 1000),
map(OfferEvent.offerDocumentSaved),
);
}

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, Event } from '@marblejs/core';
+import { matchEvent, use } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import { createMicroservice, messagingListener, Transport, MsgEffect } from '@marblejs/messaging';
+import { createMicroservice, messagingListener, Transport, MsgEffect, reply } from '@marblejs/messaging';
import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
import { IO } from 'fp-ts/lib/IO';
@@ -17,8 +17,8 @@ const fibonacci$: MsgEffect = event$ => {
mergeMap(event => of(event).pipe(
tap(event => { if (event.payload >= 45) throw new Error('Too high number!') }),
map(event => fib(event.payload)),
- map(payload => ({ ...event, type: 'FIB_RESULT', payload } as Event)),
- catchError(error => of({ ...event, error: { name: error.name, message: error.message } } as Event)),
+ map(payload => reply(event)({ type: 'FIB_RESULT', payload })),
+ catchError(error => of(reply(event)({ type: 'FIB_ERROR', error: { name: error.name, message: error.message } }))),
)),
);
};
@@ -38,7 +38,7 @@ const timeout$: MsgEffect = event$ => {
matchEvent('TIMEOUT'),
mergeMap(event => of(event).pipe(
delay(180 * 1000),
- mapTo({ ...event, type: 'TIMEOUT_RESULT' } as Event),
+ mapTo(reply(event)({ type: 'TIMEOUT_RESULT' })),
)),
);
};

ADD
packages/messaging/src/effects/messaging.effects.helper.ts
+import { Event, EventMetadata } from '@marblejs/core';
+import { isString, ExtendableError } from '@marblejs/core/dist/+internal/utils';
+
+export class MissingEventTypeError extends ExtendableError {
+ constructor() {
+ super('MissingEventTypeError', `#reply - Missing type literal`);
+ }
+}
+
+function assertEventType(event: Partial<Event>): asserts event is Required<Event> {
+ if (!event.type) throw new MissingEventTypeError();
+}
+
+function isEventMetadata(metadata: any): metadata is EventMetadata {
+ return metadata.correlationId || metadata.replyTo;
+}
+
+export const reply = (to: string | EventMetadata | Event) => <T extends Event>(event: T): T => {
+ assertEventType(event);
+
+ return isString(to)
+ ? { ...event, metadata: { replyTo: to } }
+ : isEventMetadata(to)
+ ? { ...event, metadata: to }
+ : { ...to, ...event };
+};

ADD
packages/messaging/src/effects/specs/messaging.effects.helper.spec.ts
+import { Event, EventMetadata } from '@marblejs/core';
+import { reply, MissingEventTypeError } from '../messaging.effects.helper';
+
+describe('#reply', () => {
+ test('creates success response based on originated event', () => {
+ // given
+ const origin: Event = {
+ type: 'TEST',
+ metadata: { correlationId: '#123', replyTo: 'test_channel' },
+ };
+
+ // when
+ const response = reply(origin)({
+ type: 'TEST_RESPONSE',
+ payload: 'payload',
+ });
+
+ // then
+ expect(response).toEqual({
+ type: 'TEST_RESPONSE',
+ payload: 'payload',
+ metadata: origin.metadata,
+ });
+ });
+
+ test('creates error response based on originated event', () => {
+ // given
+ const origin: Event = {
+ type: 'TEST',
+ metadata: { correlationId: '#123', replyTo: 'test_channel' },
+ };
+
+ // when
+ const response = reply(origin)({
+ type: 'TEST_RESPONSE',
+ error: 'error',
+ });
+
+ // then
+ expect(response).toEqual({
+ type: 'TEST_RESPONSE',
+ error: 'error',
+ metadata: origin.metadata,
+ });
+ });
+
+ test('creates response based on provided metadata', () => {
+ // given
+ const metadata: EventMetadata = {
+ correlationId: '#123',
+ replyTo: 'test_channel',
+ };
+
+ // when
+ const response = reply(metadata)({
+ type: 'TEST_RESPONSE',
+ payload: 'payload',
+ });
+
+ // then
+ expect(response).toEqual({
+ type: 'TEST_RESPONSE',
+ payload: 'payload',
+ metadata,
+ });
+ });
+
+ test('creates response based on provided channel', () => {
+ // given
+ const replyTo = 'test_channel';
+
+ // when
+ const response = reply(replyTo)({
+ type: 'TEST_RESPONSE',
+ payload: 'payload',
+ });
+
+ // then
+ expect(response).toEqual({
+ type: 'TEST_RESPONSE',
+ payload: 'payload',
+ metadata: {
+ replyTo,
+ },
+ });
+ });
+
+ test('throws an error if event type is not provided in reply object', () => {
+ // given
+ const origin: Event = {
+ type: 'TEST',
+ metadata: { correlationId: '#123', replyTo: 'test_channel' },
+ };
+
+ // when
+ const response = () => reply(origin)({
+ payload: 'payload',
+ } as any);
+
+ // then
+ expect(response).toThrowError(MissingEventTypeError);
+ });
+});

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -10,6 +10,7 @@ export { LocalStrategyOptions, EVENT_BUS_CHANNEL } from './transport/strategies/
// effects
export * from './effects/messaging.effects.interface';
+export * from './effects/messaging.effects.helper';
// server
export * from './server/messaging.server';

feat(messaging): #reply utiltiy function