feat(websockets): log listening, connection, incoming, outgoing and error events
MODIFY
packages/@integration/src/websockets/websockets.server.ts
packages/@integration/src/websockets/websockets.server.ts
import {
WsEffect,
- WsMiddlewareEffect,
WsConnectionEffect,
WebSocketConnectionError,
webSocketListener,
createWebSocketServer,
} from '@marblejs/websockets';
import { iif, throwError, of } from 'rxjs';
-import { tap, mergeMap, buffer, map } from 'rxjs/operators';
+import { mergeMap, buffer, map } from 'rxjs/operators';
import { matchEvent, use } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
@@ -25,11 +24,6 @@ const add$: WsEffect = (event$, ctx) =>
map(payload => ({ type: 'SUM_RESULT', payload })),
);
-const logger$: WsMiddlewareEffect = event$ =>
- event$.pipe(
- tap(e => console.log(`type: ${e.type}, payload: ${e.payload}`)),
- );
-
const connection$: WsConnectionEffect = req$ =>
req$.pipe(
mergeMap(req => iif(
@@ -42,7 +36,7 @@ const connection$: WsConnectionEffect = req$ =>
export const webSocketServer = createWebSocketServer({
connection$,
webSocketListener: webSocketListener({
- middlewares: [logger$],
+ middlewares: [],
effects: [add$],
}),
});

MODIFY
packages/core/src/index.ts
packages/core/src/index.ts
@@ -15,7 +15,7 @@ export * from './http/http.interface';
// core - error
export { coreErrorFactory, CoreErrorOptions } from './error/error.factory';
-export { CoreError, EventError } from './error/error.model';
+export { CoreError, EventError, isCoreError, isEventError } from './error/error.model';
// core - effects
export { combineEffects, combineMiddlewares } from './effects/effects.combiner';

MODIFY
packages/middleware-io/test/io-ws.integration.spec.ts
packages/middleware-io/test/io-ws.integration.spec.ts
@@ -37,6 +37,7 @@ describe('@marblejs/middleware-io - WebSocket integration', () => {
const expectedError = {
type: 'POST_USER',
error: {
+ name: 'EventError',
message: 'Validation error',
data: [{ path: 'age', expected: 'number', got: '"100"' }],
},

MODIFY
packages/websockets/src/error/specs/websocket.error.effect.spec.ts
packages/websockets/src/error/specs/websocket.error.effect.spec.ts
-import { EventError } from '@marblejs/core';
import { Marbles } from '@marblejs/core/dist/+internal/testing';
import { defaultError$ } from '../websocket.error.effect';
describe('defaultError$', () => {
test('returns stream of error events for defined error object', () => {
// given
- const event = { type: 'TEST_EVENT' };
- const error = new EventError(event, 'Test error message', { errorData: 'test_error_data' });
- const incomingEvent = { event, error };
+ const error = new Error('Test error message');
const outgoingEvent = {
- type: event.type,
+ type: 'UNHANDLED_ERROR',
error: {
+ name: error.name,
message: error.message,
- data: error.data,
},
};
// then
Marbles.assertEffect(defaultError$, [
- ['--a--', { a: incomingEvent }],
+ ['--a--', { a: error }],
['--b--', { b: outgoingEvent }],
]);
});
test('returns stream of error events for undefined error object', () => {
// given
- const event = { type: 'TEST_EVENT' };
const error = undefined;
- const incomingEvent = { event, error };
- const outgoingEvent = { type: event.type, error: {} };
-
- // then
- Marbles.assertEffect(defaultError$, [
- ['--a--', { a: incomingEvent }],
- ['--b--', { b: outgoingEvent }],
- ]);
- });
-
- test('returns stream of error events for undefined event object', () => {
- // given
- const error = undefined;
- const event = undefined;
- const incomingEvent = { event, error };
- const outgoingEvent = { type: 'ERROR', error: {} };
+ const outgoingEvent = {
+ type: 'UNHANDLED_ERROR',
+ error: {},
+ };
// then
Marbles.assertEffect(defaultError$, [
- ['--a--', { a: incomingEvent }],
+ ['--a--', { a: error }],
['--b--', { b: outgoingEvent }],
]);
});

MODIFY
packages/websockets/src/error/websocket.error.effect.ts
packages/websockets/src/error/websocket.error.effect.ts
import { map } from 'rxjs/operators';
+import { Event, isEventError, EventError } from '@marblejs/core';
import { WsErrorEffect } from '../effects/websocket.effects.interface';
-export const defaultError$: WsErrorEffect = event$ =>
- event$.pipe(
- map(error => ({
+const mapError = (error: Error | undefined): Event => ({
type: 'UNHANDLED_ERROR',
+ error: {
+ name: error?.name,
+ message: error?.message,
+ },
+});
+
+const mapEventError = (error: EventError): Event => ({
+ type: error.event.type,
error: {
name: error.name,
message: error.message,
+ data: error.data,
},
- })),
+});
+
+export const defaultError$: WsErrorEffect = event$ =>
+ event$.pipe(
+ map(error => error && isEventError(error)
+ ? mapEventError(error)
+ : mapError(error)),
);

ADD
packages/websockets/src/middlewares/websockets.eventLogger.middleware.ts
+import { tap } from 'rxjs/operators';
+import { useContext, LoggerToken, LoggerLevel, LoggerTag } from '@marblejs/core';
+import { WsMiddlewareEffect } from '../effects/websocket.effects.interface';
+
+export const inputLogger$: WsMiddlewareEffect = (event$, ctx) => {
+ const logger = useContext(LoggerToken)(ctx.ask);
+
+ return event$.pipe(
+ tap(event => {
+ const message = `${event.type}, id: ${event.metadata?.correlationId}`;
+
+ logger({
+ message,
+ type: 'EVENT_IN',
+ tag: LoggerTag.WEBSOCKETS,
+ level: LoggerLevel.INFO,
+ })();
+ }),
+ );
+};
+
+export const outputLogger$: WsMiddlewareEffect = (event$, ctx) => {
+ const logger = useContext(LoggerToken)(ctx.ask);
+
+ return event$.pipe(
+ tap(event => {
+ const { error, metadata, type } = event;
+ const message = error
+ ? `"${error.name}", "${error.message}" for event ${type}`
+ : metadata?.replyTo
+ ? `${event.type}, id: ${metadata?.correlationId ?? '-'} and sent to ${metadata?.replyTo ?? '-'}`
+ : `${event.type}, id: ${metadata?.correlationId ?? '-'}`;
+
+ logger({
+ message,
+ type: 'EVENT_OUT',
+ tag: LoggerTag.WEBSOCKETS,
+ level: event.error ? LoggerLevel.ERROR : LoggerLevel.INFO,
+ })();
+ }),
+ );
+};
+
+export const errorLogger$: WsMiddlewareEffect = (event$, ctx) => {
+ const logger = useContext(LoggerToken)(ctx.ask);
+
+ return event$.pipe(
+ tap(event => {
+ const message = `"${event.error?.name ?? '-'}: ${event.error?.message ?? '-' }" for event ${event.type}`;
+
+ return logger({
+ message,
+ type: 'ERROR',
+ tag: LoggerTag.WEBSOCKETS,
+ level: LoggerLevel.ERROR,
+ })();
+ }),
+ );
+};

MODIFY
packages/websockets/src/operators/broadcast/websocket.broadcast.operator.ts
packages/websockets/src/operators/broadcast/websocket.broadcast.operator.ts
import { Observable } from 'rxjs';
import { mergeMap, map, mapTo } from 'rxjs/operators';
+import { Event } from '@marblejs/core';
import { WebSocketClientConnection } from '../../server/websocket.server.interface';
export const broadcast =
- <Input, T>(client: WebSocketClientConnection, fn: (input: Input) => T) =>
+ <Input extends Event, T extends Event>(client: WebSocketClientConnection, fn: (input: Input) => T) =>
(input$: Observable<Input>): Observable<T> =>
input$.pipe(
map(fn),

MODIFY
packages/websockets/src/response/websocket.response.handler.ts
packages/websockets/src/response/websocket.response.handler.ts
import * as WebSocket from 'ws';
import { Observable, of } from 'rxjs';
+import { Event } from '@marblejs/core';
import { EventTransformer } from '../transformer/websocket.transformer.interface';
type ClientResponseHandler =
(client: WebSocket, eventTransformer: EventTransformer<any>) =>
- <T>(response: T) => Observable<boolean>;
+ <T extends Event>(response: T) => Observable<boolean>;
type ServerResponseHandler =
(server: WebSocket.Server, eventTransformer: EventTransformer<any>) =>
- <T>(response: T) => Observable<boolean>;
+ <T extends Event>(response: T) => Observable<boolean>;
-export const handleResponse: ClientResponseHandler = (client, eventTransformer) => <T>(response: T) => {
+export const handleResponse: ClientResponseHandler = (client, eventTransformer) => <T extends Event>(response: T) => {
const encodedResponse = eventTransformer.encode(response);
client.send(encodedResponse);
return of(true);
};
-export const handleBroadcastResponse: ServerResponseHandler = (server, eventTransformer) => <T>(response: T) => {
+export const handleBroadcastResponse: ServerResponseHandler = (server, eventTransformer) => <T extends Event>(response: T) => {
server.clients.forEach(c => {
const encodedResponse = eventTransformer.encode(response);
c.send(encodedResponse);

MODIFY
packages/websockets/src/server/specs/websocket.server.spec.ts
packages/websockets/src/server/specs/websocket.server.spec.ts
-import { EventError, HttpStatus } from '@marblejs/core';
+import { HttpStatus, Event } from '@marblejs/core';
import { throwError, fromEvent, forkJoin, merge } from 'rxjs';
import { tap, map, mergeMap, first, toArray, take, mergeMapTo, mapTo } from 'rxjs/operators';
import { webSocketListener } from '../websocket.server.listener';
@@ -41,7 +41,7 @@ describe('WebSocket server', () => {
const echo$: WsEffect = (event$, { client }) => event$.pipe(
mergeMap(event => client.sendBroadcastResponse(event).pipe(mapTo(event))),
);
- const event = JSON.stringify({ type: 'EVENT', payload: 'test' });
+ const event = { type: 'EVENT', payload: 'test' } as Event;
const listener = webSocketListener({ effects: [echo$] });
const server = testBed.getServer();
const targetClient = testBed.getClient(0);
@@ -50,15 +50,15 @@ describe('WebSocket server', () => {
const app = await createWebSocketServer({ options: { server }, webSocketListener: listener });
await app();
- targetClient.on('open', () => targetClient.send(event));
+ targetClient.on('open', () => targetClient.send(JSON.stringify(event)));
// then
const client1$ = fromEvent(testBed.getClient(0), 'message').pipe(first());
const client2$ = fromEvent(testBed.getClient(1), 'message').pipe(first());
forkJoin(client1$, client2$).subscribe(([ message1, message2 ]: [any, any]) => {
- expect(message1.data).toEqual(event);
- expect(message2.data).toEqual(event);
+ expect(JSON.parse(message1.data)).toEqual(expect.objectContaining(event));
+ expect(JSON.parse(message2.data)).toEqual(expect.objectContaining(event));
done();
});
});
@@ -122,8 +122,8 @@ describe('WebSocket server', () => {
// given
const incomingEvent = '{ some: wrong JSON object }';
const outgoingEvent = JSON.stringify({
- type: 'ERROR',
- error: { message: 'Unexpected token s in JSON at position 2' },
+ type: 'UNHANDLED_ERROR',
+ error: { name: 'SyntaxError', message: 'Unexpected token s in JSON at position 2' },
});
const targetClient = testBed.getClient(0);
const server = testBed.getServer();
@@ -151,10 +151,13 @@ describe('WebSocket server', () => {
test('passes error (thrown by effect) through stream multiple times', async done => {
// given
const incomingEvent = JSON.stringify({ type: 'EVENT' });
- const outgoingEvent = JSON.stringify({ type: 'EVENT', error: { message: 'test message' } });
- const effect$: WsEffect = event$ => event$.pipe(
- mergeMap(event => throwError(new EventError(event, 'test message'))),
+ const outgoingEvent = JSON.stringify({ type: 'UNHANDLED_ERROR', error: { name: 'Error', message: 'test_message' } });
+
+ const effect$: WsEffect = event$ =>
+ event$.pipe(
+ mergeMapTo(throwError(new Error('test_message'))),
);
+
const targetClient = testBed.getClient(0);
const server = testBed.getServer();
const listener = webSocketListener({ effects: [effect$] });
@@ -230,7 +233,7 @@ describe('WebSocket server', () => {
});
});
- describe('binary transformer', () => {
+ describe.only('binary transformer', () => {
const testBed = createWebSocketsTestBed();
beforeEach(testBed.bootstrap);
@@ -239,14 +242,18 @@ describe('WebSocket server', () => {
test('operates over binary events', async done => {
// given
const targetClient = testBed.getClient();
- const decodedMessage = 'hello world';
- const eventTransformer: EventTransformer<any, Buffer> = {
- decode: event => event,
- encode: event => event,
+ const message = 'hello world';
+
+ const eventTransformer: EventTransformer<Buffer> = {
+ decode: event => ({ type: 'BUFFER_EVENT', payload: event }),
+ encode: event => event.payload as Buffer,
};
- const effect$: WsEffect<Buffer, string> = event$ => event$.pipe(
- map(event => event.toString('utf8'))
+
+ const effect$: WsEffect<Event<Buffer>, Event<Buffer>> = event$ =>
+ event$.pipe(
+ map(event => ({ ...event, payload: Buffer.from(message) })),
);
+
const server = testBed.getServer();
const listener = webSocketListener({ effects: [effect$], eventTransformer });
@@ -255,12 +262,12 @@ describe('WebSocket server', () => {
await app();
targetClient.once('open', () => {
- targetClient.send(Buffer.from(decodedMessage));
+ targetClient.send(Buffer.from(message));
});
// then
- targetClient.once('message', incomingMessage => {
- expect(incomingMessage).toEqual(decodedMessage);
+ targetClient.once('message', (incomingMessage: Buffer) => {
+ expect(incomingMessage.toString('utf8')).toEqual(message);
done();
});
});

MODIFY
packages/websockets/src/server/websocket.server.helper.ts
packages/websockets/src/server/websocket.server.helper.ts
import * as WebSocket from 'ws';
import { EMPTY, fromEvent, merge, SchedulerLike, throwError, interval, from, iif, of } from 'rxjs';
import { takeUntil, catchError, tap, mapTo, timeout, map, mergeMap, mergeMapTo } from 'rxjs/operators';
-import { WebSocketStatus, WebSocketConnectionLiveness, WebSocketData } from '../websocket.interface';
+import { WebSocketStatus, WebSocketConnectionLiveness } from '../websocket.interface';
import { WebSocketConnectionError } from '../error/websocket.error.model';
import { EventTransformer } from '../transformer/websocket.transformer.interface';
import { handleBroadcastResponse } from '../response/websocket.response.handler';
@@ -10,7 +10,7 @@ import { WebSocketServerConnection, WebSocketClientConnection } from './websocke
export const HEART_BEAT_INTERVAL = 10 * 1000;
export const HEART_BEAT_TERMINATE_INTERVAL = HEART_BEAT_INTERVAL + 1000;
-export const createServer = (options: WebSocket.ServerOptions, eventTransformer: EventTransformer<any, WebSocketData>) => {
+export const createServer = (options: WebSocket.ServerOptions, eventTransformer: EventTransformer<any>) => {
const server = new WebSocket.Server(options) as WebSocketServerConnection;
server.sendBroadcastResponse = handleBroadcastResponse(server, eventTransformer);
return server;

MODIFY
packages/websockets/src/server/websocket.server.interface.ts
packages/websockets/src/server/websocket.server.interface.ts
import * as WebSocket from 'ws';
import { Observable } from 'rxjs';
-import { BoundDependency } from '@marblejs/core';
+import { BoundDependency, Event } from '@marblejs/core';
import { WsConnectionEffect } from '../effects/websocket.effects.interface';
import { webSocketListener } from './websocket.server.listener';
@@ -12,11 +12,13 @@ export interface WebSocketServerConfig {
}
export interface WebSocketServerConnection extends WebSocket.Server {
- sendBroadcastResponse: <T>(response: T) => Observable<boolean>;
+ sendBroadcastResponse: <T extends Event>(response: T) => Observable<boolean>;
}
export interface WebSocketClientConnection extends WebSocket {
+ id: string;
+ address: string;
isAlive: boolean;
- sendResponse: <T>(response: T) => Observable<boolean>;
- sendBroadcastResponse: <T>(response: T) => Observable<boolean>;
+ sendResponse: <T extends Event>(response: T) => Observable<boolean>;
+ sendBroadcastResponse: <T extends Event>(response: T) => Observable<boolean>;
}

MODIFY
packages/websockets/src/server/websocket.server.listener.ts
packages/websockets/src/server/websocket.server.listener.ts
@@ -18,19 +18,20 @@ import { WsEffect, WsErrorEffect, WsMiddlewareEffect, WsOutputEffect } from '../
import { jsonTransformer } from '../transformer/websocket.json.transformer';
import { EventTransformer } from '../transformer/websocket.transformer.interface';
import { defaultError$ } from '../error/websocket.error.effect';
+import { inputLogger$, outputLogger$, errorLogger$ } from '../middlewares/websockets.eventLogger.middleware';
import { WebSocketClientConnection } from './websocket.server.interface';
export interface WebSocketListenerConfig {
effects?: WsEffect<any, any>[];
middlewares?: WsMiddlewareEffect<any, any>[];
error$?: WsErrorEffect;
- eventTransformer?: EventTransformer<Event, any>;
+ eventTransformer?: EventTransformer<any>;
output$?: WsOutputEffect;
}
export interface WebSocketListener {
(connection: WebSocketClientConnection): void;
- eventTransformer: EventTransformer<Event, any>;
+ eventTransformer: EventTransformer<any>;
}
const defaultEffect$: WsEffect = msg$ => msg$;
@@ -42,18 +43,18 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
effects = [defaultEffect$],
error$ = defaultError$,
output$ = defaultOutput$,
- eventTransformer = jsonTransformer as EventTransformer<Event, any>,
+ eventTransformer = jsonTransformer,
} = config ?? {};
const logger = useContext(LoggerToken)(ask);
- const combinedMiddlewares = combineMiddlewares(...middlewares); // @TODO: create inputLogger$
+ const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
const combinedEffects = combineEffects(...effects);
const processError$ = (ctx: EffectContext<WebSocketClientConnection>) => (error: Error) =>
pipe(
of(error),
e$ => error$(e$, ctx),
- // e$ => errorLogger$(e$, ctx), // @TODO
+ e$ => errorLogger$(e$, ctx),
);
const handle = (client: WebSocketClientConnection) => {
@@ -62,9 +63,18 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
const close$ = fromEvent(client, 'close');
const message$ = fromEvent<MessageEvent>(client, 'message');
+ const applyMetadata = (event: Event): Event => ({
+ ...event,
+ metadata: {
+ replyTo: client.address,
+ correlationId: client.id,
+ },
+ });
+
const incomingEvent$ = pipe(
message$,
- e$ => e$.pipe(map(e => eventTransformer.decode(e.data))),
+ e$ => e$.pipe(map(msg => eventTransformer.decode(msg.data))),
+ e$ => e$.pipe(map(applyMetadata)),
e$ => combinedMiddlewares(e$, ctx),
e$ => e$.pipe(catchError(processError$(ctx))),
);
@@ -73,7 +83,10 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
eventSubject.asObservable(),
e$ => combinedEffects(e$, ctx),
e$ => output$(e$, ctx),
- // e$ => outputLogger$(e$, ctx), // @TODO
+ e$ => e$.pipe(map(applyMetadata)),
+ e$ => outputLogger$(e$, ctx),
+ // eslint-disable-next-line @typescript-eslint/no-unused-vars
+ e$ => e$.pipe(map(({ metadata, ...event }) => event)),
e$ => e$.pipe(catchError(processError$(ctx))),
);

MODIFY
packages/websockets/src/transformer/websocket.json.transformer.ts
packages/websockets/src/transformer/websocket.json.transformer.ts
-import { Event } from '@marblejs/core';
import { EventTransformer } from './websocket.transformer.interface';
-export const jsonTransformer: EventTransformer<Event, string> = {
+export const jsonTransformer: EventTransformer<string> = {
decode: event => JSON.parse(event),
encode: event => JSON.stringify(event),
};

MODIFY
packages/websockets/src/transformer/websocket.transformer.interface.ts
packages/websockets/src/transformer/websocket.transformer.interface.ts
+import { Event } from '@marblejs/core';
import { WebSocketData } from '../websocket.interface';
-export interface EventTransformer<T, U extends WebSocketData = WebSocketData> {
- decode: (incomingEvent: U) => T;
- encode: (outgoingEvent: T) => U;
+export interface EventTransformer<U extends WebSocketData = WebSocketData> {
+ decode: (incomingEvent: U) => Event;
+ encode: (outgoingEvent: Event) => U;
}

feat(websockets): log listening, connection, incoming, outgoing and error events