imp(messaging): effect error handling
MODIFY
packages/messaging/src/effects/messaging.effects.interface.ts
packages/messaging/src/effects/messaging.effects.interface.ts
@@ -10,7 +10,7 @@ export interface MsgMiddlewareEffect<
export interface MsgErrorEffect<
Err extends Error = Error,
-> extends MsgEffect<Err, any, MsgClient, Err> {}
+> extends MsgEffect<Event, Event, MsgClient, Err> {}
export interface MsgEffect<
I = Event,

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -5,8 +5,8 @@ import {
combineEffects,
createEffectMetadata,
} from '@marblejs/core';
-import { Observable, Subscription, Subject, merge } from 'rxjs';
-import { map, publish, withLatestFrom, takeUntil } from 'rxjs/operators';
+import { Observable, Subscription, Subject, of } from 'rxjs';
+import { map, publish, withLatestFrom, takeUntil, catchError, mergeMap } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -42,6 +42,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const message$ = conn.consumeMessage().pipe(
map(msg => ({ ...msg, data: msgTransformer.decode(msg.data) } as TransportMessage<any>)),
+ mergeMap(msg => of(msg).pipe(
publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), conn, defaultMetadata).pipe(
withLatestFrom(msg$),
)),
@@ -50,16 +51,12 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
+ catchError(error => error$(of(msg.data), conn, createEffectMetadata({ ask, error })).pipe(
+ map(data => ({ ...msg, data } as TransportMessage<any>)),
+ )),
+ ))
);
- error$(merge(
- errorSubject.asObservable(),
- conn.error$,
- ), conn, defaultMetadata)
- .pipe(takeUntil(conn.close$))
- .subscribe();
-
-
const onSubscribeEffectsOutput = (conn: TransportLayerConnection) => (msg: TransportMessage<any>) => {
if (msg.replyTo) {
conn.emitMessage(msg.replyTo, {

MODIFY
packages/messaging/src/server/messaging.server.spec.ts
packages/messaging/src/server/messaging.server.spec.ts
-import { matchEvent, use } from '@marblejs/core';
+import { Event, matchEvent, use } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import { map, tap } from 'rxjs/operators';
+import { map, tap, mergeMapTo } 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';
+import { MsgEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
+import { Subject, throwError } from 'rxjs';
describe('messagingServer', () => {
@@ -17,14 +17,16 @@ describe('messagingServer', () => {
queueOptions: { durable: false },
};
- const microservice = (effect?: MsgEffect) => createMicroservice({
+ const runServer = (effect$?: MsgEffect, error$?: MsgErrorEffect) =>
+ createMicroservice({
options,
transport: Transport.AMQP,
- messagingListener: messagingListener(effect ? { effects: [effect] } : undefined),
- });
+ messagingListener: messagingListener(effect$ ? { effects: [effect$], error$ } : undefined),
+ }).run();
+
+ const runClient = () =>
+ createAmqpStrategy(options).connect();
- const runServer = (effect?: MsgEffect) => microservice(effect).run();
- const runClient = () => createAmqpStrategy(options).connect();
const createMessage = (data: any): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
});
@@ -52,6 +54,7 @@ describe('messagingServer', () => {
test('emits event to consumer', async done => {
const eventSubject = new Subject();
+
const event$: MsgEffect = event$ =>
event$.pipe(
matchEvent('EVENT_TEST'),
@@ -72,6 +75,34 @@ describe('messagingServer', () => {
setTimeout(() => server.close().then(done), 1000);
});
});
+
+ test('reacts to thrown error', async done => {
+ const expectedMessage = { type: 'EVENT_TEST' };
+ const expectedError = new Error('test_error');
+ const errorSubject = new Subject<[Event, Error | undefined]>();
+
+ const event$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('EVENT_TEST'),
+ mergeMapTo(throwError(expectedError)),
+ );
+
+ const error$: MsgErrorEffect = (event$, _, { error }) =>
+ event$.pipe(
+ tap(e => errorSubject.next([e, error])),
+ );
+
+ const server = await runServer(event$, error$);
+ const client = await runClient();
+
+ await client.emitMessage(options.queue, createMessage(expectedMessage));
+
+ errorSubject.subscribe(data => {
+ expect(data[0]).toEqual(expectedMessage);
+ expect(data[1]).toEqual(expectedError);
+ setTimeout(() => server.close().then(done), 1000);
+ });
+ });
});
});

imp(messaging): effect error handling