feat(messaging): don't skip error events
Server doesn't have to skip error events since act operator maps default errors to `*_UNHANDLED_ERROR` prefixed event types
MODIFY
packages/core/src/operators/act/act.operator.spec.ts
packages/core/src/operators/act/act.operator.spec.ts
@@ -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 event_error: Event = { type: 'TEST_EVENT_2', error, metadata: { replyTo: 'channel_2' } };
+ const event_error: Event = { type: 'TEST_EVENT_2_UNHANDLED_ERROR', error, metadata: { replyTo: 'channel_2' } };
// when
const effect$ = (event$: Observable<Event>) =>
@@ -219,20 +219,20 @@ describe('#act operator', () => {
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' };
+ const event_ok: Event = { type: 'TEST_EVENT_PROCESSED', payload: 0 };
+ const event_err: Event = { type: 'TEST_EVENT_2_UNHANDLED_ERROR', error: 'some_error' };
// when
const effect$ = (event$: Observable<Event>) =>
event$.pipe(
- act(_ => of(eventP)),
+ act(_ => of(event_ok)),
);
// 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 }],
+ ['-a-b-c-d---', { a: event_ok, b: event_err, c: event_ok, d: event_ok }],
]);
});
@@ -279,7 +279,7 @@ describe('#act operator', () => {
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' });
+ expect(event).toEqual({ type: 'TEST_EVENT_2_UNHANDLED_ERROR', error: 'test_error' });
return event_err;
}),
);

MODIFY
packages/core/src/operators/act/act.operator.ts
packages/core/src/operators/act/act.operator.ts
@@ -30,8 +30,12 @@ export function act<
errorFn?: (error: any, event: InputEvent) => ErrorEvent | Observable<ErrorEvent>,
) {
+ const DEFAULT_ERROR_SUFFIX = '_UNHANDLED_ERROR';
+
const getDefaultErrorEvent = (error: any) => (event: Event) => of({
- type: event.type,
+ type: !event.type.includes(DEFAULT_ERROR_SUFFIX)
+ ? event.type + DEFAULT_ERROR_SUFFIX
+ : event.type,
error: encodeError(error) ?? true,
metadata: event.metadata,
} as ErrorEvent);

MODIFY
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
import { tap } from 'rxjs/operators';
import { useContext, LoggerToken, LoggerLevel } from '@marblejs/core';
+import { isNamedError } from '@marblejs/core/dist/+internal/utils';
import { MsgMiddlewareEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken } from '../server/messaging.server.tokens';
@@ -32,7 +33,7 @@ export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
const { error, metadata, type } = event;
const tag = transportLayer.config.channel;
const message = error
- ? `"${error.name}", "${error.message}" for event ${type}`
+ ? `${isNamedError(error) ? `"${error.name}", "${error.message}"` : 'error'} for event ${type}`
: metadata?.replyTo
? `${event.type}, id: ${metadata?.correlationId || '-'} and sent to "${metadata.replyTo || '-'}"`
: metadata?.correlationId

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -11,7 +11,7 @@ import {
} from '@marblejs/core';
import { pipe } from 'fp-ts/lib/pipeable';
import { Observable, Subject, defer } from 'rxjs';
-import { map, catchError, takeUntil, filter } from 'rxjs/operators';
+import { map, catchError, takeUntil } from 'rxjs/operators';
import { TransportMessageTransformer, TransportLayerConnection } from '../transport/transport.interface';
import { jsonTransformer, decodeMessage } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
@@ -77,7 +77,6 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const incomingEvent$ = pipe(
connection.message$,
map(decode),
- filter(event => !event.error),
e$ => combinedMiddlewares(e$, ctx),
e$ => defer(() => processError(e$)),
);

feat(messaging): don't skip error events

Server doesn't have to skip error events since act operator maps default errors to `*_UNHANDLED_ERROR` prefixed event types