refactor(messaging): adjusted package to changed Effect interface + moved initiator to output$
MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
@@ -36,7 +36,7 @@ const rootValiadtor$ = requestValidator$({
const test$ = r.pipe(
r.matchPath('/test'),
r.matchType('GET'),
- r.useEffect((req$, _, { ask }) => req$.pipe(
+ r.useEffect((req$, { ask }) => req$.pipe(
mergeMap(() => useContext(ClientToken)(ask).emit({ type: 'TEST' })),
mapTo(({ body: 'OK' })),
)),
@@ -45,7 +45,7 @@ const test$ = r.pipe(
const fib$ = r.pipe(
r.matchPath('/fib/:number'),
r.matchType('GET'),
- r.useEffect((req$, _, { ask }) => req$.pipe(
+ r.useEffect((req$, { ask }) => req$.pipe(
use(rootValiadtor$),
map(req => Number(req.params.number)),
mergeMap(number => {

MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -57,12 +57,13 @@ const error$: MsgServerEffect = event$ =>
tap(({ error }) => console.error(error)),
);
-const output$: MsgOutputEffect = (event$, _, { initiator }) =>
+const output$: MsgOutputEffect = event$ =>
event$.pipe(
- tap(event => console.log(`processed ::`, event, initiator && {
+ tap(({ event, initiator }) => console.log(`processed ::`, event, {
replyTo: initiator.replyTo,
correlationId: initiator.correlationId,
})),
+ map(({ event }) => event),
);
export const microservice = createMicroservice({

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 } from '../transport/transport.interface';
+import { TransportLayerConnection, TransportMessage } from '../transport/transport.interface';
type MsgClient = TransportLayerConnection;
@@ -22,7 +22,7 @@ export interface MsgEffect<
export interface MsgOutputEffect<
I = Event,
O = Event,
-> extends MsgEffect<I, O> {}
+> extends MsgEffect<{ event: I; initiator: TransportMessage<any> }, O> {}
export interface MsgServerEffect<T extends Event = Event>
extends MsgEffect<T, any, undefined> {}

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

MODIFY
packages/messaging/src/server/messaging.server.ts
packages/messaging/src/server/messaging.server.ts
@@ -24,8 +24,8 @@ export const createMicroservice = (config: CreateMicroserviceConfig) => {
if (event$) {
const serverEvent$ = serverEventSubject.asObservable().pipe(takeWhile(e => !isCloseEvent(e)));
- const metadata = createEffectMetadata({ ask: lookup(context) });
- event$(serverEvent$, undefined, metadata).subscribe();
+ const metadata = createEffectMetadata({ ask: lookup(context), client: undefined });
+ event$(serverEvent$, metadata).subscribe();
}
return listenerWithContext.listen;

refactor(messaging): adjusted package to changed Effect interface + moved initiator to output$