feat(websockets): detect closed connections
MODIFY
packages/@integration/src/websockets/websockets.server.ts
packages/@integration/src/websockets/websockets.server.ts
-import {
- WsEffect,
- WsConnectionEffect,
- WebSocketConnectionError,
- webSocketListener,
- createWebSocketServer,
-} from '@marblejs/websockets';
-import { iif, throwError, of } from 'rxjs';
-import { mergeMap, buffer, map } from 'rxjs/operators';
+import { WsEffect, webSocketListener, createWebSocketServer } from '@marblejs/websockets';
+import { buffer, map } from 'rxjs/operators';
import { matchEvent, use } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
@@ -24,25 +17,9 @@ const add$: WsEffect = (event$, ctx) =>
map(payload => ({ type: 'SUM_RESULT', payload })),
);
-const error$: WsEffect = event$ =>
- event$.pipe(
- matchEvent('ERROR'),
- mergeMap(() => throwError(new Error('some_error_message'))),
- );
-
-const connection$: WsConnectionEffect = req$ =>
- req$.pipe(
- mergeMap(req => iif(
- () => req.headers.upgrade !== 'websocket',
- throwError(new WebSocketConnectionError('Unauthorized', 4000)),
- of(req),
- )),
- );
-
export const webSocketServer = createWebSocketServer({
- connection$,
listener: webSocketListener({
middlewares: [],
- effects: [add$, error$],
+ effects: [add$],
}),
});

MODIFY
packages/websockets/src/index.ts
packages/websockets/src/index.ts
// server
export { webSocketListener, WebSocketListenerConfig } from './server/websocket.server.listener';
export { createWebSocketServer } from './server/websocket.server';
+export * from './server/websocket.server.event';
export * from './server/websocket.server.interface';
// common

MODIFY
packages/websockets/src/middlewares/websockets.statusLogger.middleware.ts
packages/websockets/src/middlewares/websockets.statusLogger.middleware.ts
@@ -38,13 +38,27 @@ const close$: WsServerEffect = (event$, ctx) => {
return event$.pipe(
matchEvent(ServerEvent.close),
- map(event => event.payload),
- tap(logger({
+ tap(() => logger({
type: 'Server',
message: 'Server connection was closed',
level: LoggerLevel.INFO,
tag: LoggerTag.WEBSOCKETS,
- }))
+ })())
+ );
+};
+
+const closeClient$: WsServerEffect = (event$, ctx) => {
+ const logger = useContext(LoggerToken)(ctx.ask);
+
+ return event$.pipe(
+ matchEvent(ServerEvent.closeClient),
+ map(event => event.payload),
+ tap(({ client }) => logger({
+ type: 'Server',
+ message: `Closed connection form client "${client.id}" (${client.address})`,
+ level: LoggerLevel.INFO,
+ tag: LoggerTag.WEBSOCKETS,
+ })())
);
};
@@ -68,4 +82,5 @@ export const statusLogger$ = combineEffects(
listening$,
error$,
close$,
+ closeClient$,
);

MODIFY
packages/websockets/src/server/websocket.server.event.ts
packages/websockets/src/server/websocket.server.event.ts
@@ -7,6 +7,7 @@ export enum ServerEventType {
HEADERS = 'headers',
CONNECTION = 'connection',
CLOSE = 'close',
+ CLOSE_CLIENT = 'close_client',
ERROR = 'error',
}
@@ -26,6 +27,10 @@ export const ServerEvent = {
close: createEvent(
ServerEventType.CLOSE,
),
+ closeClient: createEvent(
+ ServerEventType.CLOSE_CLIENT,
+ (client: WebSocketClientConnection) => ({ client }),
+ ),
error: createEvent(
ServerEventType.ERROR,
(error: Error) => ({ error }),
@@ -50,6 +55,10 @@ export function isCloseEvent(event: Event): event is ReturnType<typeof ServerEve
return event.type === ServerEventType.CLOSE;
}
+export function isCloseClientEvent(event: Event): event is ReturnType<typeof ServerEvent.closeClient> {
+ return event.type === ServerEventType.CLOSE_CLIENT;
+}
+
export function isErrorEvent(event: Event): event is ReturnType<typeof ServerEvent.error> {
return event.type === ServerEventType.ERROR;
}

MODIFY
packages/websockets/src/server/websocket.server.ts
packages/websockets/src/server/websocket.server.ts
@@ -76,7 +76,8 @@ export const createWebSocketServer = async (config: WebSocketServerConfig) => {
const listen: ServerIO<WebSocketServerConnection> = () => new Promise((resolve, reject) => {
handleServerBrokenConnections(server).subscribe();
- server.once(ServerEventType.CONNECTION, (client: WebSocketClientConnection, req: http.IncomingMessage) => {
+ server.on(ServerEventType.CONNECTION, (client: WebSocketClientConnection, req: http.IncomingMessage) => {
+ client.once('close', () => serverEventSubject.next(ServerEvent.closeClient(client)));
client.sendResponse = handleResponse(client, webSocketListener.eventTransformer);
client.sendBroadcastResponse = handleBroadcastResponse(server, webSocketListener.eventTransformer);
client.isAlive = true;

feat(websockets): detect closed connections