refactor(messaging): pass error to stream instead of metadata atrribute
MODIFY
packages/messaging/src/effects/messaging.effects.interface.ts
packages/messaging/src/effects/messaging.effects.interface.ts
import { Effect, Event } from '@marblejs/core';
-import { TransportLayerConnection, TransportMessage } from '../transport/transport.interface';
+import { TransportLayerConnection } from '../transport/transport.interface';
type MsgClient = TransportLayerConnection;
@@ -9,15 +9,15 @@ export interface MsgMiddlewareEffect<
> extends MsgEffect<I, O> {}
export interface MsgErrorEffect<
+ I = Event,
Err extends Error = Error,
-> extends MsgEffect<Event, Event, MsgClient, Err> {}
+> extends MsgEffect<{ event: I; error: Err }, Event, MsgClient> {}
export interface MsgEffect<
I = Event,
O = Event,
Client = MsgClient,
- Err extends Error = Error,
-> extends Effect<I, O, Client, Err, TransportMessage<any>> {}
+> extends Effect<I, O, Client> {}
export interface MsgOutputEffect<
I = Event,

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -30,12 +30,15 @@ export interface MessagingListenerConfig {
msgTransformer?: TransportMessageTransformer<any>;
}
+const defaultOutput$: MsgOutputEffect = msg$ => msg$;
+const defaultError$: MsgErrorEffect = msg$ => msg$.pipe(map(m => m.event));
+
export const messagingListener = (config: MessagingListenerConfig = {}) => {
const {
effects = [],
middlewares = [],
- output$ = (msg$ => msg$) as MsgOutputEffect,
- error$ = (msg$ => msg$) as MsgErrorEffect,
+ output$ = defaultOutput$,
+ error$ = defaultError$,
msgTransformer = jsonTransformer,
} = config;
@@ -49,23 +52,24 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const errorSubject = new Subject<Error>();
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(...middlewares);
+ const metadata = createEffectMetadata({ ask });
const message$ = conn.message$.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, createEffectMetadata({ ask, initiator: msg })).pipe(
+ publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), conn, metadata).pipe(
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- publish(msg$ => combinedEffects(msg$.pipe(map(m => m.data)), conn, createEffectMetadata({ ask, initiator: msg })).pipe(
+ publish(msg$ => combinedEffects(msg$.pipe(map(m => m.data)), conn, metadata).pipe(
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- publish(msg$ => output$(msg$.pipe(map(m => m.data)), conn, createEffectMetadata({ ask, initiator: msg })).pipe(
+ publish(msg$ => output$(msg$.pipe(map(m => m.data)), conn, metadata).pipe(
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- catchError(error => error$(of(msg.data), conn, createEffectMetadata({ ask, error })).pipe(
+ catchError(error => error$(of({ event: msg.data, error }), conn, metadata).pipe(
map(data => ({ ...msg, data } as TransportMessage<any>)),
)),
))

MODIFY
packages/messaging/src/server/messaging.server.spec.ts
packages/messaging/src/server/messaging.server.spec.ts
@@ -95,9 +95,10 @@ describe('messagingServer', () => {
mergeMapTo(throwError(expectedError)),
);
- const error$: MsgErrorEffect = (event$, _, { error }) =>
+ const error$: MsgErrorEffect = event$ =>
event$.pipe(
- tap(e => errorSubject.next([e, error])),
+ tap(({ event, error }) => errorSubject.next([event, error])),
+ map(({ event }) => event),
);
const server = await runServer(event$, error$);

refactor(messaging): pass error to stream instead of metadata atrribute