fix(messaging): decodeMessage - pass metadata from raw message
MODIFY
packages/messaging/src/transport/transport.transformer.spec.ts
packages/messaging/src/transport/transport.transformer.spec.ts
import { ReplaySubject } from 'rxjs';
import { take } from 'rxjs/operators';
+import { Event } from '@marblejs/core';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { TransportMessage } from './transport.interface';
import { jsonTransformer, decodeMessage } from './transport.transformer';
@@ -54,13 +55,23 @@ describe('#decodeMessage', () => {
// given
const message: TransportMessage<Buffer> = {
data: Buffer.from('{ type }'),
+ correlationId: createUuid(),
+ replyTo: createUuid(),
+ raw: {},
};
// when
const decodedEvent = decodeMessage({ errorSubject, msgTransformer })(message);
// then
- expect(decodedEvent).toEqual({ type: 'UNKNOWN' });
+ expect(decodedEvent).toEqual({
+ type: 'UNKNOWN',
+ metadata: expect.objectContaining({
+ correlationId: message.correlationId,
+ replyTo: message.replyTo,
+ raw: expect.anything(),
+ })
+ } as Event);
errorSubject
.pipe(take(1))

MODIFY
packages/messaging/src/transport/transport.transformer.ts
packages/messaging/src/transport/transport.transformer.ts
@@ -8,28 +8,28 @@ import { TransportMessageTransformer, TransportMessage } from './transport.inter
export type DecodeMessageConfig = { msgTransformer: TransportMessageTransformer; errorSubject: Subject<Error> };
export type DecodeMessage = (config: DecodeMessageConfig) => (msg: TransportMessage<Buffer>) => Event;
-const UNKNOWN_EVENT = { type: 'UNKNOWN' };
-
export const jsonTransformer: TransportMessageTransformer = {
decode: event => JSON.parse(event.toString()),
encode: event => flow(JSON.stringify, Buffer.from)(event),
};
+const applyMetadata = (raw: TransportMessage<Buffer>) => (event: Event) => ({
+ ...event,
+ metadata: {
+ replyTo: raw.replyTo,
+ correlationId: raw.correlationId,
+ raw,
+ },
+});
+
export const decodeMessage: DecodeMessage = ({ msgTransformer, errorSubject }) => msg =>
pipe(
E.tryCatch(
() => msgTransformer.decode(msg.data),
error => {
errorSubject.next(error as Error);
- return UNKNOWN_EVENT;
+ return applyMetadata(msg)({ type: 'UNKNOWN' });
}),
- E.map(event => ({
- ...event,
- metadata: {
- replyTo: msg.replyTo,
- correlationId: msg.correlationId,
- raw: msg,
- },
- })),
+ E.map(applyMetadata(msg)),
E.fold(identity, identity),
);

fix(messaging): decodeMessage - pass metadata from raw message