fix(messaging): map unhandled errors to events
MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
@@ -45,22 +45,22 @@ const test$ = r.pipe(
const fib$ = r.pipe(
r.matchPath('/fib/:number'),
r.matchType('GET'),
- r.useEffect((req$, { ask }) => req$.pipe(
+ r.useEffect((req$, ctx) => {
+ const client = useContext(ClientToken)(ctx.ask);
+
+ return req$.pipe(
use(rootValiadtor$),
map(req => Number(req.params.number)),
- mergeMap(number => {
- const client = useContext(ClientToken)(ask);
-
- return forkJoin(
+ mergeMap(number => forkJoin(
client.send({ type: 'FIB', payload: number + 0 }),
client.send({ type: 'FIB', payload: number + 1 }),
client.send({ type: 'FIB', payload: number + 2 }),
client.send({ type: 'FIB', payload: number + 3 }),
client.send({ type: 'FIB', payload: number + 4 }),
+ )),
+ map(body => ({ body })),
);
}),
- map(body => ({ body })),
- )),
);
const listening$: HttpServerEffect = event$ =>

MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -27,7 +27,6 @@ export const microservice = createMicroservice({
}),
});
-
if (process.env.NODE_ENV !== 'test') {
microservice();
}

MODIFY
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
@@ -32,7 +32,9 @@ export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
return event$.pipe(
tap(({ event, initiator }) => {
- const eventType = chalk.yellow(event.type);
+ const isError = !!event.error;
+ const colorize = isError ? chalk.red : chalk.yellow;
+ const eventType = colorize(event.type);
const eventCorrelationId = initiator.correlationId && chalk.magenta(initiator.correlationId);
const eventReplyTo = initiator.replyTo && chalk.yellow(initiator.replyTo);
const channel = transportLayer.config.channel;
@@ -41,7 +43,7 @@ export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
: `${eventType}, id: ${eventCorrelationId}`;
logger({
- level: LoggerLevel.INFO,
+ level: isError ? LoggerLevel.ERROR : LoggerLevel.INFO,
tag: 'EVENT_OUT',
message,
channel,

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -11,8 +11,9 @@ import {
import * as O from 'fp-ts/lib/Option';
import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
-import { Observable, Subscription, Subject, of, zip } from 'rxjs';
-import { map, publish, takeUntil, catchError, take } from 'rxjs/operators';
+import { flow } from 'fp-ts/lib/function';
+import { Observable, Subscription, Subject, of, zip, OperatorFunction} from 'rxjs';
+import { map, publish, takeUntil, catchError, take, mergeMapTo } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -25,6 +26,8 @@ import { TransportLayerToken, ServerEventsToken } from './messaging.server.token
import { AllServerEvents, ServerEvent } from './messaging.server.events';
import { inputLogger$, outputLogger$, errorLogger$ } from '../middlewares/messaging.eventLogger.middleware';
+type ProcessOperator = OperatorFunction<TransportMessage<any>, TransportMessage<any>>;
+
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
middlewares?: MsgMiddlewareEffect<any, any>[];
@@ -56,35 +59,65 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const errorSubject = new Subject<Error>();
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
- const combinedOutput = combineEffects(outputLogger$, output$);
- const combinedError = combineEffects(errorLogger$, error$);
const ctx = createEffectContext({ ask, client: conn });
+ const toUnhandledErrorEvent = ({ name, message }: Error): Observable<Event> =>
+ of({ type: 'UNHANDLED_ERROR', error: { name, message } });
+
const decode = (msg: TransportMessage<Buffer>): TransportMessage<Event> => ({
...msg,
data: { ...msgTransformer.decode(msg.data), raw: msg },
});
- const message$ = conn.message$.pipe(
- map(decode),
+ const processMiddlewares: ProcessOperator = flow(
publish(msg$ => zip(
- combinedMiddlewares(msg$.pipe(map(m => m.data)), ctx),
+ combinedMiddlewares(msg$.pipe(map(m => m.data)), ctx).pipe(
+ catchError(toUnhandledErrorEvent),
+ ),
msg$,
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
+ );
+
+ const processEffects: ProcessOperator = flow(
publish(msg$ => zip(
- combinedEffects(msg$.pipe(map(m => m.data)), ctx),
+ combinedEffects(msg$.pipe(map(m => m.data)), ctx).pipe(
+ catchError(toUnhandledErrorEvent),
+ ),
msg$,
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
+ );
+
+ const processOutput: ProcessOperator = flow(
publish(msg$ => zip(
- combinedOutput(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx),
+ outputLogger$(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx).pipe(
+ catchError(toUnhandledErrorEvent),
+ ),
msg$,
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- catchError((error: EventError) => combinedError(of({ event: error.event, error }), ctx).pipe(
- map(data => ({ data } as TransportMessage<any>)),
+ publish(msg$ => zip(
+ output$(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx).pipe(
+ catchError(toUnhandledErrorEvent),
+ ),
+ msg$,
)),
+ map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
+ );
+
+ const message$ = conn.message$.pipe(
+ map(decode),
+ processMiddlewares,
+ processEffects,
+ processOutput,
+ catchError((error: EventError) => {
+ const e$ = of({ event: error.event, error });
+ return errorLogger$(e$, ctx).pipe(
+ mergeMapTo(error$(e$, ctx)),
+ map(data => ({ data } as TransportMessage<any>)),
+ );
+ }),
);
const onSubscribeEffectsOutput = (conn: TransportLayerConnection) => (msg: TransportMessage<any>) => {

MODIFY
packages/messaging/src/server/messaging.server.logger.ts
packages/messaging/src/server/messaging.server.logger.ts
@@ -4,13 +4,15 @@ import { getOrElse } from 'fp-ts/lib/Option';
import { ContextProvider } from '@marblejs/core';
import { LoggerToken } from './messaging.server.tokens';
+const IS_TEST_ENV = process.env.NODE_ENV === 'test';
+
export enum LoggerLevel { INFO, WARN, ERROR }
export type LoggerOptions = { channel: string; message: string; tag: string; level?: LoggerLevel };
export type Logger = (opts: LoggerOptions) => any;
export const provideLogger = (ask: ContextProvider): Logger => pipe(
ask(LoggerToken),
- getOrElse(() => defaultLogger),
+ getOrElse(() => !IS_TEST_ENV ? defaultLogger : () => undefined),
);
const defaultLogger: Logger = ({ channel, message, tag, level = LoggerLevel.INFO }) => {

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
-import { Event, matchEvent, use, EventError, createEvent } from '@marblejs/core';
+import { Event, matchEvent, use, combineEffects } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
+import { Subject, throwError } from 'rxjs';
import { map, tap, mergeMapTo, delay, bufferCount } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
-import { MsgEffect, MsgErrorEffect } from '../../effects/messaging.effects.interface';
-import { Subject, throwError } from 'rxjs';
+import { MsgEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
import { AmqpStrategyOptions } from '../../transport/strategies/amqp.strategy.interface';
import { runClient, runServer, createMessage } from '../../util/messaging.test.util';
@@ -98,69 +98,44 @@ describe('messagingServer::AMQP', () => {
});
test('reacts to thrown Error and doesn\'t crash internal messages stream', async done => {
- const event = { type: 'EVENT_TEST' };
+ const event1: Event = { type: 'EVENT_TEST_1' };
+ const event2: Event = { type: 'EVENT_TEST_2' };
const error = new Error('test_error');
- const errorSubject = new Subject<[Event | undefined, Error]>();
+ const expectedError = { error: { message: error.message, name: error.name }, type: 'UNHANDLED_ERROR' };
+ const outputSubject = new Subject<Event>();
- errorSubject
+ outputSubject
.pipe(bufferCount(2))
.subscribe(data => {
- expect(data[0][0]).toBeUndefined();
- expect(data[0][1]).toEqual(error);
- expect(data[1][0]).toBeUndefined();
- expect(data[1][1]).toEqual(error);
+ expect(data[0].type).toEqual(expectedError.type);
+ expect(data[0].error).toEqual(expectedError.error);
+ expect(data[1].type).toEqual(event2.type);
+ expect(data[1].payload).toEqual(event2.payload);
setTimeout(() => server.close().then(done), 1000);
});
- const event$: MsgEffect = event$ =>
+ const event1$: MsgEffect = event$ =>
event$.pipe(
- matchEvent('EVENT_TEST'),
+ matchEvent(event1.type),
mergeMapTo(throwError(error)),
);
- const error$: MsgErrorEffect = event$ =>
+ const event2$: MsgEffect = event$ =>
event$.pipe(
- tap(({ event, error }) => errorSubject.next([event, error])),
- map(() => createEvent('ERROR')),
+ matchEvent(event2.type),
);
- const options = createOptions({ expectAck: false });
- const client = await runClient(Transport.AMQP, options);
- const server = await runServer(Transport.AMQP, options)(event$, error$);
-
- await client.emitMessage(options.queue, createMessage(event));
- await client.emitMessage(options.queue, createMessage(event));
- });
-
- test('reacts to thrown EventError', async done => {
- const event = { type: 'EVENT_TEST' };
- const error = new EventError(event, 'test_error');
- const errorSubject = new Subject<[Event | undefined, Error]>();
-
- errorSubject.subscribe(data => {
- expect(data[0]).toEqual(event);
- expect(data[1]).toEqual(error);
- setTimeout(() => server.close().then(done), 1000);
- });
-
- const event$: MsgEffect = event$ =>
- event$.pipe(
- matchEvent('EVENT_TEST'),
- mergeMapTo(throwError(error)),
- );
-
- const error$: MsgErrorEffect = event$ =>
+ const output$: MsgOutputEffect = event$ =>
event$.pipe(
- tap(({ event, error }) => errorSubject.next([event, error])),
- map(() => createEvent('ERROR')),
+ tap(({ event }) => outputSubject.next(event)),
+ map(({ event }) => event),
);
const options = createOptions({ expectAck: false });
const client = await runClient(Transport.AMQP, options);
- const server = await runServer(Transport.AMQP, options)(event$, error$);
+ const server = await runServer(Transport.AMQP, options)(combineEffects(event1$, event2$), output$);
- const emitResult = await client.emitMessage(options.queue, createMessage(event));
-
- expect(emitResult).toEqual(true);
+ 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, createEvent } from '@marblejs/core';
+import { matchEvent, use } from '@marblejs/core';
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 { MsgEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
import { RedisStrategyOptions } from '../../transport/strategies/redis.strategy.interface';
import { runClient, runServer, createMessage, wait } from '../../util/messaging.test.util';
@@ -74,28 +74,34 @@ describe('messagingServer::Redis', () => {
});
test('throws an UnsupportedError for unsupported "ackMessage/nackMessage"', async done => {
+ const expectedEventType = 'UNHANDLED_ERROR';
+ const expectedEventError = {
+ name: 'UnsupportedError',
+ message: 'Unsupported operation. Method \"ackMessage\" is unsupported for Redis transport layer.',
+ };
+
const event$: MsgEffect = (event$, ctx) =>
event$.pipe(
matchEvent('EVENT_TEST'),
tap(event => ctx.client.ackMessage(event.raw)),
);
- const error$: MsgErrorEffect = event$ =>
+ const output$: MsgOutputEffect = event$ =>
event$.pipe(
- tap(async ({ event, error }) => {
- expect(event).toBeUndefined();
- expect(error).toEqual(error);
+ tap(async ({ event }) => {
+ expect(event.type).toEqual(expectedEventType);
+ expect(event.error).toEqual(expectedEventError);
await wait();
await server.close();
await client.close();
done();
}),
- map(() => createEvent('ERROR')),
+ map(({ event }) => event),
);
const options = createOptions();
const client = await runClient(Transport.REDIS, options);
- const server = await runServer(Transport.REDIS, options)(event$, error$);
+ const server = await runServer(Transport.REDIS, options)(event$, output$);
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 { createMicroservice } from '../server/messaging.server';
import { TransportMessage, Transport } from '../transport/transport.interface';
-import { MsgEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
+import { MsgEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { messagingListener } from '../server/messaging.server.listener';
import { provideTransportLayer } from '../transport/transport.provider';
-export const runServer = (transport: any, options: any) => (effect$?: MsgEffect, error$?: MsgErrorEffect) =>
+export const runServer = (transport: any, options: any) => (effect$?: MsgEffect, output$?: MsgOutputEffect) =>
createMicroservice({
options,
transport,
- messagingListener: messagingListener(effect$ ? { effects: [effect$], error$ } : undefined),
+ messagingListener: messagingListener(effect$ ? { effects: [effect$], output$ } : undefined),
})();
export const runClient = (transport: Transport, transportOptions: any) =>

fix(messaging): map unhandled errors to events