feat(core): act - pass through error events
MODIFY
packages/core/src/+internal/utils/error.util.ts
packages/core/src/+internal/utils/error.util.ts
@@ -9,3 +9,11 @@ export const isNamedError = (data: any): data is NamedError =>
export const isError = (data: any): data is Error =>
!!data?.stack && !!data?.name;
+
+export const encodeError = (error: any) =>
+ !isError(error) ? error : ['name', ...Object.getOwnPropertyNames(error)]
+ .filter(key => !['stack'].includes(key))
+ .reduce((acc, key) => {
+ acc[key] = error[key];
+ return acc;
+ }, Object.create(null));

MODIFY
packages/core/src/operators/act/act.operator.spec.ts
packages/core/src/operators/act/act.operator.spec.ts
-import { Observable, of, concat } from 'rxjs';
+import { Observable, of, concat, throwError } from 'rxjs';
import { filter, tap, mergeMap } from 'rxjs/operators';
import { Event } from '../../event/event.interface';
import { Marbles } from '../../+internal/testing';
@@ -79,7 +79,7 @@ describe('#act operator', () => {
const event2: Event = { type: 'TEST_EVENT_2', payload: 2, metadata: { replyTo: 'channel_2' } };
const event3: Event = { type: 'TEST_EVENT_3', payload: 3, metadata: { replyTo: 'channel_3' } };
const event4: Event = { type: 'TEST_EVENT_4', payload: 4, metadata: { replyTo: 'channel_4' } };
- const eventErrored: Event = { type: 'TEST_EVENT_2', error, metadata: { replyTo: 'channel_2' } };
+ const event_error: Event = { type: 'TEST_EVENT_2', error, metadata: { replyTo: 'channel_2' } };
// when
const effect$ = (event$: Observable<Event>) =>
@@ -92,7 +92,7 @@ describe('#act operator', () => {
// then
Marbles.assertEffect(effect$, [
['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
- ['-a-b-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event1, b: event_error, c: event3, d: event4 }],
]);
});
@@ -102,7 +102,7 @@ describe('#act operator', () => {
const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
- const eventErrored: Event = { type: 'TEST_ERROR', error: 'something went wrong' };
+ const event_error: Event = { type: 'TEST_ERROR', error: 'something went wrong' };
// when
const effect$ = (event$: Observable<Event>) =>
@@ -121,7 +121,7 @@ describe('#act operator', () => {
// then
Marbles.assertEffect(effect$, [
['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
- ['-a-b-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event1, b: event_error, c: event3, d: event4 }],
]);
});
@@ -131,7 +131,7 @@ describe('#act operator', () => {
const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
- const eventErrored: Event = { type: 'TEST_EVENT_2', error: 'something went wrong' };
+ const event_error: Event = { type: 'TEST_EVENT_2', error: 'something went wrong' };
// when
const effect$ = (event$: Observable<Event>) =>
@@ -150,7 +150,7 @@ describe('#act operator', () => {
// then
Marbles.assertEffect(effect$, [
['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
- ['-a-b-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event1, b: event_error, c: event3, d: event4 }],
]);
});
@@ -160,7 +160,7 @@ describe('#act operator', () => {
const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
- const eventErrored: Event = { type: 'TEST_ERROR' };
+ const event_error: Event = { type: 'TEST_ERROR' };
// when
const effect$ = (event$: Observable<Event>) =>
@@ -170,8 +170,8 @@ describe('#act operator', () => {
tap(e => { if (e.payload === 2) throw new Error('something went wrong') }),
),
() => concat(
- of(eventErrored),
- of(eventErrored),
+ of(event_error),
+ of(event_error),
),
),
);
@@ -179,7 +179,7 @@ describe('#act operator', () => {
// then
Marbles.assertEffect(effect$, [
['-a-b----c-d---', { a: event1, b: event2, c: event3, d: event4 }],
- ['-a-(bb)-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ['-a-(bb)-c-d---', { a: event1, b: { ...event_error, error: true }, c: event3, d: event4 }],
]);
});
@@ -189,7 +189,7 @@ describe('#act operator', () => {
const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
- const eventErrored: Event = { type: 'TEST_ERROR' };
+ const event_error: Event = { type: 'TEST_ERROR' };
// when
const effect$ = (event$: Observable<Event>) =>
@@ -200,8 +200,8 @@ describe('#act operator', () => {
else return of(event);
},
() => concat(
- of(eventErrored),
- of(eventErrored),
+ of(event_error),
+ of(event_error),
),
),
);
@@ -209,7 +209,85 @@ describe('#act operator', () => {
// then
Marbles.assertEffect(effect$, [
['-a-b----c-d---', { a: event1, b: event2, c: event3, d: event4 }],
- ['-a-(bb)-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ['-a-(bb)-c-d---', { a: event1, b: { ...event_error, error: true }, c: event3, d: event4 }],
+ ]);
+ });
+
+ test('passes through errored event', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', error: 'some_error' };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+ const eventP: Event = { type: 'TEST_EVENT_PROCESSED', payload: 0 };
+
+ const eventE: Event = { type: 'TEST_EVENT_2', error: 'some_error' };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act(_ => of(eventP)),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: eventP, b: eventE, c: eventP, d: eventP }],
+ ]);
+ });
+
+ test('chains "act" operators', () => {
+ // given
+ const event1: Event<number> = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event<number, string> = { type: 'TEST_EVENT_2', error: 'some_error' };
+ const event3: Event<number> = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event<number> = { type: 'TEST_EVENT_4', payload: 4 };
+
+ const event_ok: Event<boolean> = { type: 'TEST_EVENT_PROCESSED', payload: true };
+ const event_ok_out: Event<number> = { type: 'TEST_EVENT_PROCESSED', payload: 0 };
+
+ const event_err1: Event<string> = { type: 'TEST_EVENT_ERROR', payload: 'test_1' };
+ const event_err2: Event<string> = { type: 'TEST_EVENT_ERROR', payload: 'test_2' };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act(_ => of(event_ok), _ => event_err1),
+ act(_ => of(event_ok_out), _ => event_err2),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event_ok_out, b: { ...event_err2, error: true }, c: event_ok_out, d: event_ok_out }],
+ ]);
+ });
+
+ test('triggers subsequent "act" error handler if previous "act" operator did not handled the exception', () => {
+ // given
+ const event1: Event<number> = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event<number> = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event<number> = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event<number> = { type: 'TEST_EVENT_4', payload: 4 };
+
+ const event_ok: Event<boolean> = { type: 'TEST_EVENT_PROCESSED', payload: true };
+ const event_err: Event<string> = { type: 'TEST_EVENT_ERROR', payload: 'test_error' };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act(e => e.payload === 2 ? throwError('test_error') : of(e as Event<number>)),
+ act(_ => of(event_ok), (error, event) => {
+ expect(error).toEqual('test_error');
+ expect(event).toEqual({ type: 'TEST_EVENT_2', error: 'test_error' });
+ return event_err;
+ }),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event_ok, b: { ...event_err, error: true }, c: event_ok, d: event_ok }],
]);
});
});

MODIFY
packages/core/src/operators/act/act.operator.ts
packages/core/src/operators/act/act.operator.ts
import * as O from 'fp-ts/lib/Option';
import { pipe } from 'fp-ts/lib/pipeable';
import { Observable, of, defer, isObservable } from 'rxjs';
-import { mergeMap, catchError } from 'rxjs/operators';
+import { mergeMap, catchError, map } from 'rxjs/operators';
+import { encodeError } from '../../+internal/utils';
import { Event } from '../../event/event.interface';
export function act<
@@ -18,7 +19,7 @@ export function act<
>(
callFn: (event: InputEvent) => Observable<CallEvent>,
errorFn: (error: any, event: InputEvent) => ErrorEvent | Observable<ErrorEvent>,
-): (source: Observable<InputEvent>) => Observable<CallEvent | ErrorEvent>;
+): (source: Observable<InputEvent>) => Observable<CallEvent>;
export function act<
InputEvent extends Event,
@@ -31,22 +32,25 @@ export function act<
const getDefaultErrorEvent = (error: any) => (event: Event) => of({
type: event.type,
- error: { name: error.name, message: error.message, data: error.data },
+ error: encodeError(error) ?? true,
metadata: event.metadata,
} as ErrorEvent);
- const handleError = (event: InputEvent) => (error: unknown) => pipe(
+ const handleError = (event: InputEvent) => (error: unknown): Observable<any> => pipe(
O.fromNullable(errorFn),
O.map(fn => fn(error, event)),
O.map(res => !isObservable(res) ? of(res) : res),
+ O.map(res => res.pipe(map(r => ({ ...r, error: r.error ?? true })))),
O.getOrElse(() => getDefaultErrorEvent(error)(event)),
);
- return (source: Observable<InputEvent>): Observable<CallEvent | ErrorEvent> =>
+ return (source: Observable<InputEvent>): Observable<CallEvent> =>
source.pipe(
mergeMap(event => defer(() => {
try {
- return pipe(
+ return event.error
+ ? handleError(event)(event.error)
+ : pipe(
callFn(event),
catchError(handleError(event)));
} catch (error) {

MODIFY
packages/messaging/src/middlewares/messaging.eventOutput.middleware.ts
packages/messaging/src/middlewares/messaging.eventOutput.middleware.ts
import { Event, useContext } from '@marblejs/core';
-import { createUuid, isNonNullable } from '@marblejs/core/dist/+internal/utils';
+import { createUuid, isNonNullable, encodeError } from '@marblejs/core/dist/+internal/utils';
import { map } from 'rxjs/operators';
import { MsgOutputEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken } from '../server/messaging.server.tokens';
-export const encodeError = (error: any) =>
- ['name', ...Object.getOwnPropertyNames(error)].reduce((acc, key) => {
- acc[key] = error[key];
- return acc;
- }, Object.create(null))
-
export const outputRouter$: MsgOutputEffect = (event$, ctx) => {
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
const originChannel = transportLayer.config.channel;

feat(core): act - pass through error events