feat(messaging): try to set `correlationId` to each incoming event
ADD
packages/messaging/src/middlewares/messaging.eventInput.middleware.ts
+import { map } from 'rxjs/operators';
+import { createUuid } from '@marblejs/core/dist/+internal/utils';
+import { MsgMiddlewareEffect } from '../effects/messaging.effects.interface';
+
+export const idApplier$: MsgMiddlewareEffect = event$ =>
+ event$.pipe(
+ map(event => ({
+ ...event,
+ metadata: {
+ ...event.metadata,
+ correlationId: event.metadata?.correlationId ?? createUuid(),
+ },
+ }))
+ );

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -17,6 +17,7 @@ import { jsonTransformer, decodeMessage } from '../transport/transport.transform
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { inputLogger$, outputLogger$, errorLogger$ } from '../middlewares/messaging.eventLogger.middleware';
import { outputRouter$, outputErrorEncoder$ } from '../middlewares/messaging.eventOutput.middleware';
+import { idApplier$ } from '../middlewares/messaging.eventInput.middleware';
export interface MessagingListenerConfig {
effects?: MsgEffect[];
@@ -52,7 +53,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const logger = useContext(LoggerToken)(ask);
const combinedEffects = combineEffects(...effects);
- const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
+ const combinedMiddlewares = combineMiddlewares(idApplier$, inputLogger$, ...middlewares);
return connection => {
const errorSubject = new Subject<Error>();

feat(messaging): try to set `correlationId` to each incoming event