fix(messaging): map to UNKNOWN event if event decoder fails
MODIFY
packages/messaging/src/client/messaging.client.interface.ts
packages/messaging/src/client/messaging.client.interface.ts
@@ -3,13 +3,13 @@ import { Observable } from 'rxjs';
import { TransportMessageTransformer, TransportStrategy } from '../transport/transport.interface';
export interface MessagingClient {
- send: <T = Event, U = Event>(data: U) => Observable<T>;
- emit: <T = Event>(data: T) => Promise<void>;
+ send: <T extends Event = Event, U extends Event = Event>(data: U) => Observable<T>;
+ emit: <T extends Event = Event>(data: T) => Promise<void>;
close: () => Promise<void>;
}
type ConfigurationBase = {
- msgTransformer?: TransportMessageTransformer<any>;
+ msgTransformer?: TransportMessageTransformer;
}
export type MessagingClientConfig =

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -34,16 +34,16 @@ export const messagingClient = (config: MessagingClientConfig) => {
const transportLayer = provideTransportLayer(transport, options);
const connection = await transportLayer.connect({ isConsumer: false });
- const emit: MessagingClient['emit'] = async msg => {
+ const emit = async <I extends Event>(event: I): Promise<void> => {
await connection.emitMessage(
connection.getChannel(),
- { data: msgTransformer.encode(msg as any) },
+ { data: msgTransformer.encode(event) },
);
}
- const send: MessagingClient['send'] = msg => {
+ const send = <O extends Event, I extends Event>(event: I): Observable<O> => {
const channel = connection.getChannel();
- const message: TransportMessage<Buffer> = { data: msgTransformer.encode(msg as any) };
+ const message: TransportMessage<Buffer> = { data: msgTransformer.encode(event) };
const catchErrorEvent = <V extends Event>(e: V) => {
if (!e.error) {
@@ -72,13 +72,13 @@ export const messagingClient = (config: MessagingClientConfig) => {
return defer(() => from(connection.sendMessage(channel, message)).pipe(
timeout(config.options.timeout || DEFAULT_TIMEOUT),
- map(m => msgTransformer.decode(m.data)),
+ map(m => msgTransformer.decode<O>(m.data)),
mergeMap(catchErrorEvent),
take(1),
));
}
- const close: MessagingClient['close'] = async () => {
+ const close = async (): Promise<void> => {
await connection.close();
}

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -12,29 +12,34 @@ import {
import { pipe } from 'fp-ts/lib/pipeable';
import { Observable, Subject, defer } from 'rxjs';
import { map, catchError, takeUntil, filter } from 'rxjs/operators';
-import { TransportMessage, TransportMessageTransformer, TransportLayerConnection } from '../transport/transport.interface';
-import { jsonTransformer } from '../transport/transport.transformer';
+import { TransportMessageTransformer, TransportLayerConnection } from '../transport/transport.interface';
+import { jsonTransformer, decodeMessage } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { inputLogger$, outputLogger$, errorLogger$ } from '../middlewares/messaging.eventLogger.middleware';
import { outputRouter$, outputErrorEncoder$ } from '../middlewares/messaging.eventOutput.middleware';
export interface MessagingListenerConfig {
- effects?: MsgEffect<any, any>[];
- middlewares?: MsgMiddlewareEffect<any, any>[];
+ effects?: MsgEffect[];
+ middlewares?: MsgMiddlewareEffect[];
error$?: MsgErrorEffect;
output$?: MsgOutputEffect;
- msgTransformer?: TransportMessageTransformer<any>;
+ msgTransformer?: TransportMessageTransformer;
}
export interface MessagingListener {
(connection: TransportLayerConnection): void;
}
-const defaultOutput$: MsgOutputEffect = msg$ =>
- msg$;
+const defaultOutput$: MsgOutputEffect = event$ =>
+ event$;
-const defaultError$: MsgErrorEffect = msg$ =>
- msg$.pipe(map(error => ({ type: 'UNHANDLED_ERROR', error: { name: error.name, message: error.message } } as Event)));
+const defaultError$: MsgErrorEffect = event$ =>
+ event$.pipe(
+ map(error => ({
+ type: 'UNHANDLED_ERROR',
+ error: { name: error.name, message: error.message }
+ })),
+ );
export const messagingListener = createListener<MessagingListenerConfig, MessagingListener>(config => ask => {
const {
@@ -49,14 +54,11 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
- const decode = (msg: TransportMessage<Buffer>): Event => ({
- ...msgTransformer.decode(msg.data),
- metadata: {
- replyTo: msg.replyTo,
- correlationId: msg.correlationId,
- raw: msg,
- }
- });
+ return connection => {
+ const errorSubject = new Subject<Error>();
+ const eventSubject = new Subject<Event>();
+ const decode = decodeMessage({ msgTransformer, errorSubject });
+ const ctx = createEffectContext({ ask, client: connection });
const send = (connection: TransportLayerConnection) => (event: Event): void => {
const { metadata, type, payload, error } = event;
@@ -70,14 +72,10 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
});
};
- return connection => {
- const errorSubject = new Subject<Error>();
- const eventSubject = new Subject<Event>();
- const ctx = createEffectContext({ ask, client: connection });
-
const incomingEvent$ = pipe(
connection.message$,
- e$ => e$.pipe(map(decode), filter(event => !event.error)),
+ map(decode),
+ filter(event => !event.error),
e$ => combinedMiddlewares(e$, ctx),
e$ => defer(() => processError(e$)),
);

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -49,9 +49,9 @@ export interface TransportLayerConnection {
close$: Observable<any>;
}
-export interface TransportMessageTransformer<T> {
- decode: (incomingEvent: Buffer) => T;
- encode: (outgoingEvent: T) => Buffer;
+export interface TransportMessageTransformer {
+ decode: <T extends Event = Event>(incomingEvent: Buffer) => T;
+ encode: (outgoingEvent: Event) => Buffer;
}
export interface TransportMessage<T = Event> {

ADD
packages/messaging/src/transport/transport.transformer.spec.ts
+import { ReplaySubject } from 'rxjs';
+import { take } from 'rxjs/operators';
+import { createUuid } from '@marblejs/core/dist/+internal/utils';
+import { TransportMessage } from './transport.interface';
+import { jsonTransformer, decodeMessage } from './transport.transformer';
+
+describe('#decodeMessage', () => {
+ const errorSubject = new ReplaySubject<Error>(1);
+
+ describe('#jsonTransformer', () => {
+ const msgTransformer = jsonTransformer;
+
+ test('decodes JSON message to Event with routing metadata', () => {
+ // given
+ const incomingEvent = { type: 'TEST', payload: { test: 'test' } };
+ const message: TransportMessage<Buffer> = {
+ data: msgTransformer.encode(incomingEvent),
+ correlationId: createUuid(),
+ replyTo: createUuid(),
+ };
+
+ // when
+ const decodedEvent = decodeMessage({ errorSubject, msgTransformer })(message);
+
+ // then
+ expect(decodedEvent).toEqual({
+ ...incomingEvent,
+ metadata: {
+ correlationId: message.correlationId,
+ replyTo: message.replyTo,
+ raw: message,
+ },
+ });
+ });
+
+ test('decodes JSON message to Event without routing metadata', () => {
+ // given
+ const incomingEvent = { type: 'TEST', payload: { test: 'test' } };
+ const message: TransportMessage<Buffer> = {
+ data: msgTransformer.encode(incomingEvent),
+ };
+
+ // when
+ const decodedEvent = decodeMessage({ errorSubject, msgTransformer })(message);
+
+ // then
+ expect(decodedEvent).toEqual({
+ ...incomingEvent,
+ metadata: { raw: message },
+ });
+ });
+
+ test('decodes invalid JSON message to unknown Event and emits error to given subject', done => {
+ // given
+ const message: TransportMessage<Buffer> = {
+ data: Buffer.from('{ type }'),
+ };
+
+ // when
+ const decodedEvent = decodeMessage({ errorSubject, msgTransformer })(message);
+
+ // then
+ expect(decodedEvent).toEqual({ type: 'UNKNOWN' });
+
+ errorSubject
+ .pipe(take(1))
+ .subscribe(error => {
+ expect(error).toBeDefined();
+ done();
+ });
+ });
+ });
+});

MODIFY
packages/messaging/src/transport/transport.transformer.ts
packages/messaging/src/transport/transport.transformer.ts
+import { Subject } from 'rxjs';
+import { flow, identity } from 'fp-ts/lib/function';
+import { pipe } from 'fp-ts/lib/pipeable';
+import * as E from 'fp-ts/lib/Either';
import { Event } from '@marblejs/core';
-import { flow } from 'fp-ts/lib/function';
-import { TransportMessageTransformer } from './transport.interface';
+import { TransportMessageTransformer, TransportMessage } from './transport.interface';
-export const jsonTransformer: TransportMessageTransformer<Event> = {
+export type DecodeMessageConfig = { msgTransformer: TransportMessageTransformer; errorSubject: Subject<Error> };
+export type DecodeMessage = (config: DecodeMessageConfig) => (msg: TransportMessage<Buffer>) => Event;
+
+const UNKNOWN_EVENT = { type: 'UNKNOWN' };
+
+export const jsonTransformer: TransportMessageTransformer = {
decode: event => JSON.parse(event.toString()),
encode: event => flow(JSON.stringify, Buffer.from)(event),
};
+
+export const decodeMessage: DecodeMessage = ({ msgTransformer, errorSubject }) => msg =>
+ pipe(
+ E.tryCatch(
+ () => msgTransformer.decode(msg.data),
+ error => {
+ errorSubject.next(error as Error);
+ return UNKNOWN_EVENT;
+ }),
+ E.map(event => ({
+ ...event,
+ metadata: {
+ replyTo: msg.replyTo,
+ correlationId: msg.correlationId,
+ raw: msg,
+ },
+ })),
+ E.fold(identity, identity),
+ );

fix(messaging): map to UNKNOWN event if event decoder fails