feat(messaging): detailed logs for event validation errors
MODIFY
packages/core/src/event/event.interface.ts
packages/core/src/event/event.interface.ts
@@ -13,7 +13,6 @@ export interface Event<P = unknown, E = any, T extends EventType = EventType> {
metadata?: EventMetadata;
}
-
export interface EventWithoutPayload<T extends EventType = EventType> {
type: T;
metadata?: EventMetadata;

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
@@ -12,11 +12,15 @@ export interface EventBusConfig {
timeout?: number;
}
-export const EventBusToken = createContextToken<TransportLayerConnection<Transport.LOCAL>>('EventBus');
+export interface EventBus extends TransportLayerConnection<Transport.LOCAL> {
+ context: Context;
+}
+
+export const EventBusToken = createContextToken<EventBus>('EventBus');
export const eventBus = (config: EventBusConfig) => pipe(
R.ask<Context>(),
- R.map(async derivedContext => {
+ R.map<Context, Promise<EventBus>>(async derivedContext => {
const { listener } = config;
const transportLayer = createLocalStrategy(config);
const transportLayerConnection = await transportLayer.connect();
@@ -31,6 +35,10 @@ export const eventBus = (config: EventBusConfig) => pipe(
listener(context)(transportLayerConnection);
- return transportLayerConnection;
+ const eventBus = transportLayerConnection as EventBus;
+
+ eventBus.context = context;
+
+ return eventBus;
}),
);

ADD
packages/messaging/src/middlewares/messaging.eventLogger.middleware.spec.ts
+import {
+ LoggerToken,
+ LoggerLevel,
+ lookup,
+ bindEagerlyTo,
+ createEffectContext,
+ contextFactory,
+ createReader,
+ EffectContext,
+ EventError,
+} from '@marblejs/core';
+import { Marbles } from '@marblejs/core/dist/+internal/testing';
+import { constUndefined } from 'fp-ts/lib/function';
+import { pipe } from 'fp-ts/lib/pipeable';
+import { io } from 'fp-ts/lib/IO';
+import { messagingListener } from '../server/messaging.server.listener';
+import { eventBus, EventBus } from '../eventbus/messaging.eventBus.reader';
+import { inputLogger$, outputLogger$, exceptionLogger$ } from './messaging.eventLogger.middleware';
+
+describe('inputLogger$', () => {
+ let loggerMock: jest.Mock;
+ let ctx: EffectContext<EventBus>;
+
+ const provideLoggerMock = () =>
+ jest.fn(io.of(constUndefined));
+
+ const provideEventBus = async (logger: any) => pipe(
+ await contextFactory(bindEagerlyTo(LoggerToken)(createReader(() => logger))),
+ eventBus({ listener: messagingListener() }));
+
+ const provideEffectContextForLogger = async (logger: any) => pipe(
+ await provideEventBus(logger),
+ client => createEffectContext({ ask: lookup(client.context), client }));
+
+ beforeEach(async () => {
+ loggerMock = provideLoggerMock();
+ ctx = await provideEffectContextForLogger(loggerMock);
+ });
+
+ afterEach(() => {
+ loggerMock.mockReset();
+ })
+
+ test('logs incoming events', async () => {
+ // given - events
+ const events = [
+ { type: 'TEST_EVENT_TYPE_0' },
+ { type: 'TEST_EVENT_TYPE_1', payload: { test: true } },
+ { type: 'TEST_EVENT_TYPE_2', error: { test: true } },
+ ];
+
+ loggerMock.mockClear();
+
+ // when, then
+ Marbles.assertEffect(inputLogger$, [
+ ['-012-', { 0: events[0], 1: events[1], 2: events[2] }],
+ ['-012-', { 0: events[0], 1: events[1], 2: events[2] }],
+ ], { ctx });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.INFO,
+ message: events[0].type,
+ type: 'EVENT_IN',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.INFO,
+ message: events[1].type,
+ type: 'EVENT_IN',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.INFO,
+ message: events[2].type,
+ type: 'EVENT_IN',
+ tag: 'event_bus',
+ });
+ });
+
+ test('logs outgoing events', async () => {
+ // given - events
+ const events = [
+ {
+ type: 'TEST_EVENT_TYPE_0'
+ },
+ {
+ type: 'TEST_EVENT_TYPE_1',
+ payload: { test: true },
+ metadata: { correlationId: 'some_id', replyTo: 'some_channel' },
+ },
+ {
+ type: 'TEST_EVENT_TYPE_2',
+ error: { test: true },
+ },
+ {
+ type: 'TEST_EVENT_TYPE_3',
+ error: new Error('some_error_message'),
+ },
+ {
+ type: 'TEST_EVENT_TYPE_4',
+ error: new EventError({ type: 'TEST_EVENT_TYPE_4' }, 'some_error_message', { test: true }),
+ },
+ {
+ type: 'TEST_EVENT_TYPE_5',
+ error: new EventError({ type: 'TEST_EVENT_TYPE_5' }, 'some_error_message', { test: true }),
+ metadata: { correlationId: 'some_id' },
+ },
+ ];
+
+ loggerMock.mockClear();
+
+ // when, then
+ Marbles.assertEffect(outputLogger$, [
+ ['-012345-', { 0: events[0], 1: events[1], 2: events[2], 3: events[3], 4: events[4], 5: events[5] }],
+ ['-012345-', { 0: events[0], 1: events[1], 2: events[2], 3: events[3], 4: events[4], 5: events[5] }],
+ ], { ctx });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.INFO,
+ message: events[0].type,
+ type: 'EVENT_OUT',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.INFO,
+ message: `${events[1].type}, id: ${events[1].metadata?.correlationId} and sent to \"${events[1].metadata?.replyTo}\"`,
+ type: 'EVENT_OUT',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.ERROR,
+ message: `Unknown error "{"test":true}" for event "${events[2].type}"`,
+ type: 'EVENT_OUT',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.ERROR,
+ message: `"Error: some_error_message" for event "${events[3].type}"`,
+ type: 'EVENT_OUT',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.ERROR,
+ message: `Received invalid event "${events[4].type}" (UNKNOWN_CORRELATION_ID) with error: {"test":true}`,
+ type: 'EVENT_OUT',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.ERROR,
+ message: `Received invalid event "${events[5].type}" (some_id) with error: {"test":true}`,
+ type: 'EVENT_OUT',
+ tag: 'event_bus',
+ });
+ });
+
+ test('logs server exceptions', async () => {
+ // given - events
+ const events = [
+ { type: 'TEST_EVENT_TYPE_0' },
+ { type: 'TEST_EVENT_TYPE_1', error: new Error('some_error_message') },
+ { type: 'TEST_EVENT_TYPE_2', error: { test: true } },
+ ];
+
+ loggerMock.mockClear();
+
+ // when, then
+ Marbles.assertEffect(exceptionLogger$, [
+ ['-012-', { 0: events[0], 1: events[1], 2: events[2] }],
+ ['-012-', { 0: events[0], 1: events[1], 2: events[2] }],
+ ], { ctx });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.ERROR,
+ message: `Unknown error "undefined" for event "${events[0].type}"`,
+ type: 'ERROR',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.ERROR,
+ message: `"Error: some_error_message" for event "${events[1].type}"`,
+ type: 'ERROR',
+ tag: 'event_bus',
+ });
+
+ expect(loggerMock).toHaveBeenCalledWith({
+ level: LoggerLevel.ERROR,
+ message: `Unknown error "{"test":true}" for event "${events[2].type}"`,
+ type: 'ERROR',
+ tag: 'event_bus',
+ });
+ });
+});

MODIFY
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
+import { pipe } from 'fp-ts/lib/pipeable';
import { tap } from 'rxjs/operators';
-import { useContext, LoggerToken, LoggerLevel } from '@marblejs/core';
-import { isNamedError } from '@marblejs/core/dist/+internal/utils';
+import { useContext, LoggerToken, LoggerLevel, Event, EventError, isEventError } from '@marblejs/core';
+import { NamedError } from '@marblejs/core/dist/+internal/utils';
import { MsgMiddlewareEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken } from '../server/messaging.server.tokens';
+/**
+ * Logs incoming events
+ *
+ * @since v3.0.0
+ * @param event$ - event stream
+ * @param ctx - EffectContext
+ */
export const inputLogger$: MsgMiddlewareEffect = (event$, ctx) => {
const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
-
- return event$.pipe(
- tap(event => {
const tag = transportLayer.config.channel;
- const id = event.metadata?.correlationId;
- const message = id ? `${event.type}, id: ${id}` : event.type;
- logger({
- tag,
- message,
- type: 'EVENT_IN',
- level: LoggerLevel.INFO,
- })();
- }),
+ return event$.pipe(
+ tap(event => pipe(
+ getIncomingMessageForEvent(event),
+ message => logger({ tag, message, level: LoggerLevel.INFO, type: 'EVENT_IN' }),
+ )()),
);
};
+/**
+ * Logs outgoing events
+ *
+ * @since v3.0.0
+ * @param event$ - event stream
+ * @param ctx - EffectContext
+ */
export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
-
- return event$.pipe(
- tap(event => {
- const { error, metadata, type } = event;
const tag = transportLayer.config.channel;
- const message = error
- ? `${isNamedError(error) ? `"${error.name}", "${error.message}"` : 'error'} for event ${type}`
- : metadata?.replyTo
- ? `${event.type}, id: ${metadata?.correlationId || '-'} and sent to "${metadata.replyTo || '-'}"`
- : metadata?.correlationId
- ? `${event.type}, id: ${metadata.correlationId}`
- : event.type
- logger({
- tag,
- message,
- level: event.error ? LoggerLevel.ERROR : LoggerLevel.INFO,
- type: 'EVENT_OUT',
- })();
- }),
+ return event$.pipe(
+ tap(event => pipe(
+ event.error
+ ? getErrorMessageForEvent(event)
+ : getOutgoingMessageForEvent(event),
+ message => logger({ tag, message, level: event.error ? LoggerLevel.ERROR : LoggerLevel.INFO, type: 'EVENT_OUT' })
+ )()),
);
};
-export const errorLogger$: MsgMiddlewareEffect = (event$, ctx) => {
+/**
+ * Logs server exceptions
+ *
+ * @since v3.0.0
+ * @param event$ - event stream
+ * @param ctx - EffectContext
+ */
+export const exceptionLogger$: MsgMiddlewareEffect = (event$, ctx) => {
const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
const tag = transportLayer.config.channel;
return event$.pipe(
- tap(event => {
- const message = `"${event.error?.name || '-'}: ${event.error?.message || '-' }" for event ${event.type}`;
-
- return logger({
- tag,
- message,
- type: 'ERROR',
- level: LoggerLevel.ERROR,
- })();
- }),
+ tap(event => pipe(
+ getErrorMessageForEvent(event),
+ message => logger({ tag, message, level: LoggerLevel.ERROR, type: 'ERROR' }),
+ )()),
);
};
+
+const getValidationErrorMessage = (event: Event<unknown, EventError>): string => {
+ const error = event.error;
+ const correlationId = event.metadata?.correlationId || 'UNKNOWN_CORRELATION_ID';
+ const parsedError = error ? JSON.stringify(error.data) : 'UNKNOWN_ERROR';
+ const eventType = event.type.replace('_UNHANDLED_ERROR', '');
+ return correlationId
+ ? `Received invalid event "${eventType}" (${correlationId}) with error: ${parsedError}`
+ : `Received invalid event "${eventType}" with error: ${parsedError}`;
+}
+
+const getGenericErrorMessage = (event: Event<unknown, NamedError>): string => {
+ const error = event.error;
+ const name = error?.name;
+ const message = error?.message || '-';
+ const eventType = event.type;
+ return name
+ ? `"${name}: ${message}" for event "${eventType}"`
+ : `Unknown error "${JSON.stringify(error)}" for event "${eventType}"`
+}
+
+const getOutgoingMessageForEvent = ({ type, metadata }: Event): string => {
+ const replyTo = metadata?.replyTo;
+ const correlationId = metadata?.correlationId;
+ return replyTo && correlationId
+ ? `${type}, id: ${correlationId} and sent to "${replyTo}"`
+ : correlationId
+ ? `${type}, id: ${correlationId}`
+ : type;
+}
+
+const getIncomingMessageForEvent = ({ type, metadata }: Event): string => {
+ const id = metadata?.correlationId;
+ return id ? `${type}, id: ${id}` : type;
+};
+
+const getErrorMessageForEvent = (event: Event<unknown>): string =>
+ isEventError(event.error)
+ ? getValidationErrorMessage(event)
+ : getGenericErrorMessage(event);

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -10,12 +10,13 @@ import {
LoggerLevel,
} from '@marblejs/core';
import { pipe } from 'fp-ts/lib/pipeable';
+import { identity } from 'fp-ts/lib/function';
import { Observable, Subject, defer } from 'rxjs';
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';
-import { inputLogger$, outputLogger$, errorLogger$ } from '../middlewares/messaging.eventLogger.middleware';
+import { inputLogger$, outputLogger$, exceptionLogger$ } from '../middlewares/messaging.eventLogger.middleware';
import { outputRouter$, outputErrorEncoder$ } from '../middlewares/messaging.eventOutput.middleware';
import { idApplier$ } from '../middlewares/messaging.eventInput.middleware';
import { rejectUnhandled$ } from '../middlewares/messaging.ack.middleware';
@@ -32,9 +33,6 @@ export interface MessagingListener {
(connection: TransportLayerConnection): void;
}
-const defaultOutput$: MsgOutputEffect = event$ =>
- event$;
-
const defaultError$: MsgErrorEffect = event$ =>
event$.pipe(
map(error => ({
@@ -47,7 +45,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const {
middlewares = [],
effects = [],
- output$ = defaultOutput$,
+ output$ = identity,
error$ = defaultError$,
msgTransformer = jsonTransformer,
} = config ?? {};
@@ -94,7 +92,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const errorEvent$ = pipe(
errorSubject.asObservable(),
e$ => error$(e$, ctx),
- e$ => defer(() => errorLogger$(e$, ctx)),
+ e$ => exceptionLogger$(e$, ctx),
);
const processError = <T>(origin$: Observable<T>): Observable<T> =>

feat(messaging): detailed logs for event validation errors