fix(websockets): don't push an event to the client that is  closing
MODIFY
packages/websockets/src/+internal/websocket.test.util.ts
packages/websockets/src/+internal/websocket.test.util.ts
@@ -24,7 +24,8 @@ export const bootstrapWebSocketServer = async (
options: WebSocketServerConfig['options'],
listener: ServerConfig<any, any>['listener'],
event$?: WsServerEffect,
-) => (await createWebSocketServer({ options, listener, event$ }))();
+ dependencies?: WebSocketServerConfig['dependencies'],
+) => (await createWebSocketServer({ options, listener, event$, dependencies }))();
export const createWebSocketClientMock = (): WebSocketClientConnection => {
class WebSocketClientMock extends WebSocket.EventEmitter {

DELETE
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 extends Event>(response: T) => Observable<boolean>;
-
-type ServerResponseHandler =
- (server: WebSocket.Server, eventTransformer: EventTransformer<any>) =>
- <T extends Event>(response: T) => Observable<boolean>;
-
-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 extends Event>(response: T) => {
- server.clients.forEach(c => {
- const encodedResponse = eventTransformer.encode(response);
- c.send(encodedResponse);
- });
- return of(true);
-};

MODIFY
packages/websockets/src/server/specs/websocket.server.spec.ts
packages/websockets/src/server/specs/websocket.server.spec.ts
import { Event } from '@marblejs/core';
+import { wait } from '@marblejs/core/dist/+internal/utils';
import { throwError, fromEvent, forkJoin } from 'rxjs';
import { tap, map, mergeMap, first, toArray, take, mapTo } from 'rxjs/operators';
import { webSocketListener } from '../websocket.server.listener';
@@ -51,7 +52,7 @@ describe('WebSocket server', () => {
const client1$ = fromEvent(webSocketclient1, 'message').pipe(first());
const client2$ = fromEvent(webSocketclient2, 'message').pipe(first());
- forkJoin(client1$, client2$).subscribe(([ message1, message2 ]: [any, any]) => {
+ forkJoin([client1$, client2$]).subscribe(([ message1, message2 ]: [any, any]) => {
expect(JSON.parse(message1.data)).toEqual(event);
expect(JSON.parse(message2.data)).toEqual(event);
webSocketclient1.close();
@@ -181,6 +182,29 @@ describe('WebSocket server', () => {
done();
});
});
+
+ test('server doesn\'t try to push an event to the client that is already closing', async () => {
+ try {
+ // given - effects
+ const effect$: WsEffect = (event$, ctx) => event$.pipe(tap(_ => ctx.client.close()));
+
+ // given - server + clients
+ const listener = webSocketListener({ effects: [effect$] });
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, listener);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
+
+ // when
+ webSocketClient.send(JSON.stringify({ type: 'SOME_EVENT' }));
+ await wait(1);
+
+ // then
+ webSocketServer.close();
+ httpServer.close();
+ } catch (error) {
+ fail(`Server should not throw a fatal exception: "${error}"`);
+ }
+ });
});
describe('binary transformer', () => {

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 { Context, lookup } from '@marblejs/core';
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';
+import { broadcastEvent } from './websocket.server.response';
import { WebSocketServerConnection, WebSocketClientConnection } from './websocket.server.interface';
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>) => {
+export const createServer = (options: WebSocket.ServerOptions, eventTransformer: EventTransformer<any>, ctx: Context) => {
+ const ask = lookup(ctx);
const server = new WebSocket.Server(options) as WebSocketServerConnection;
- server.sendBroadcastResponse = handleBroadcastResponse(server, eventTransformer);
+ server.sendBroadcastResponse = broadcastEvent({ server, eventTransformer, ask });
return server;
}
+export const sendMessage = (client: WebSocket) => (outgoingEvent: any): Promise<unknown> =>
+ new Promise((res, rej) => client.send(outgoingEvent, err => err ? rej() : res()));
+
export const handleServerBrokenConnections = (server: WebSocketServerConnection, scheduler?: SchedulerLike) =>
interval(HEART_BEAT_INTERVAL, scheduler).pipe(
takeUntil(fromEvent(server, 'close')),

MODIFY
packages/websockets/src/server/websocket.server.listener.ts
packages/websockets/src/server/websocket.server.listener.ts
@@ -115,7 +115,7 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
input$
.pipe(takeUntil(close$))
.subscribe(
- (event: Event) => client.sendResponse(event),
+ (event: Event) => client.sendResponse(event).toPromise(),
(error: EventError) => {
const type = 'ServerListener';
const message = `Unexpected error for OutgoingEvent stream: "${error.name}", "${error.message}"`;

ADD
packages/websockets/src/server/websocket.server.response.ts
+import * as WebSocket from 'ws';
+import * as O from 'fp-ts/lib/Option';
+import * as T from 'fp-ts/lib/Task';
+import * as TE from 'fp-ts/lib/TaskEither';
+import { forkJoin, Observable, of } from 'rxjs';
+import { catchError, mapTo } from 'rxjs/operators';
+import { constant, constVoid, flow, pipe } from 'fp-ts/lib/function';
+import { ContextProvider, Event, LoggerToken, LoggerTag, LoggerLevel } from '@marblejs/core';
+import { EventTransformer } from '../transformer/websocket.transformer.interface';
+import { WebsocketReadyStateMap } from '../websocket.interface';
+import { sendMessage } from './websocket.server.helper';
+
+type ResponseHandlerContext = {
+ eventTransformer: EventTransformer<any>;
+ ask: ContextProvider;
+};
+
+type ClientResponseHandlerContext = ResponseHandlerContext & { client: WebSocket & { id?: string }}
+type ServerResponseHandlerContext = ResponseHandlerContext & { server: WebSocket.Server }
+
+type ClientResponseHandler = (ctx: ClientResponseHandlerContext) => <T extends Event>(event: T) => Observable<boolean>;
+type ServerResponseHandler = (ctx: ServerResponseHandlerContext) => <T extends Event>(event: T) => Observable<boolean>;
+
+const log = (level: LoggerLevel) => (ask: ContextProvider) => (message: string): T.Task<void> =>
+ pipe(
+ ask(LoggerToken),
+ O.map(logger => logger({ level, message, tag: LoggerTag.WEBSOCKETS, type: 'ServerResponseHandler' })),
+ O.getOrElse(constant(constVoid)),
+ T.fromIO);
+
+const logWarn = log(LoggerLevel.WARN);
+const logError = log(LoggerLevel.ERROR);
+
+/**
+ * Try to emit an event to given WebSocket client
+ * @TODO fix from eager to lazy evaluation (with introduction of 4.0)
+ */
+export const emitEvent: ClientResponseHandler = ({ client, eventTransformer, ask }) => <T extends Event>(event: T) => {
+ const skipSendingEvent: T.Task<boolean> =
+ pipe(
+ logWarn(ask)(`Trying to send an event for client which is ${WebsocketReadyStateMap[client.readyState]}`),
+ T.map(constant(false)),
+ );
+
+ const tryToSendEvent: T.Task<boolean> =
+ pipe(
+ TE.tryCatch(
+ () => pipe(eventTransformer.encode(event), sendMessage(client)),
+ constant(`An error occured while sending and event to client "${client.id}"`)),
+ TE.fold(
+ flow(logError(ask), T.map(constant(false))),
+ constant(T.of(true)),
+ ));
+
+ const isConnectionOpen = client.readyState === WebSocket.OPEN;
+
+ // @TODO connect it with observable with an introduction of version 4.0
+ // return pipe(
+ // defer(pipe(isConnectionOpen ? tryToSendEvent : skipSendingEvent)),
+ // catchError(constant(of(false))));
+
+ isConnectionOpen ? tryToSendEvent() : skipSendingEvent();
+
+ return of(true);
+};
+
+/**
+ * Broadcast event to all connected clients
+ */
+export const broadcastEvent: ServerResponseHandler = ({ server, eventTransformer, ask }) => <T extends Event>(event: T) => {
+
+ // @TODO connect it with observable with an introduction of version 4.0
+ // return pipe(
+ // forkJoin([...server.clients].map(client => emitEvent({ client, eventTransformer, ask })(event))),
+ // catchError(constant(of(false))),
+ // mapTo(true),
+ // );
+
+ pipe(
+ forkJoin([...server.clients].map(client => emitEvent({ client, eventTransformer, ask })(event))),
+ catchError(constant(of(false))),
+ mapTo(true),
+ ).toPromise();
+
+ return of(true);
+}

MODIFY
packages/websockets/src/server/websocket.server.ts
packages/websockets/src/server/websocket.server.ts
@@ -6,9 +6,9 @@ import { map, takeWhile } from 'rxjs/operators';
import { Context, contextFactory, createEffectContext, ServerIO, LoggerTag, lookup, logContext, combineEffects } from '@marblejs/core';
import { createUuid, isNullable } from '@marblejs/core/dist/+internal/utils';
import { createServer, handleServerBrokenConnections, handleClientBrokenConnection } from '../server/websocket.server.helper';
-import { handleBroadcastResponse, handleResponse } from '../response/websocket.response.handler';
import { WebSocketConnectionError } from '../error/websocket.error.model';
import { statusLogger$ } from '../middlewares/websockets.statusLogger.middleware';
+import { broadcastEvent, emitEvent } from './websocket.server.response';
import { WebSocketServerConfig, WebSocketClientConnection, WebSocketServerConnection } from './websocket.server.interface';
import { subscribeWebSocketEvents } from './websocket.server.event.subscriber';
import { isCloseEvent, ServerEventType, ServerEvent } from './websocket.server.event';
@@ -50,7 +50,7 @@ export const createWebSocketServer = async (config: WebSocketServerConfig) => {
verifyClient: verifyClient(context),
noServer,
...options
- }, webSocketListener.eventTransformer);
+ }, webSocketListener.eventTransformer, context);
const { serverEvent$, serverEventSubject } = subscribeWebSocketEvents(server);
@@ -63,8 +63,8 @@ export const createWebSocketServer = async (config: WebSocketServerConfig) => {
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.sendResponse = emitEvent({ client, eventTransformer: webSocketListener.eventTransformer, ask: ctx.ask });
+ client.sendBroadcastResponse = broadcastEvent({ server, eventTransformer: webSocketListener.eventTransformer, ask: ctx.ask });
client.isAlive = true;
client.id = createUuid();
client.address = req.connection.remoteAddress ?? '-';

MODIFY
packages/websockets/src/websocket.interface.ts
packages/websockets/src/websocket.interface.ts
+import * as WebSocket from 'ws';
+
export enum WebSocketStatus {
NORMAL_CLOSURE = 1000,
GOING_AWAY = 1001,
@@ -22,3 +24,10 @@ export enum WebSocketConnectionLiveness {
}
export type WebSocketData = string | Buffer | ArrayBuffer | Buffer[];
+
+export const WebsocketReadyStateMap = {
+ [WebSocket.OPEN]: 'open',
+ [WebSocket.CLOSED]: 'closed',
+ [WebSocket.CLOSING]: 'closing',
+ [WebSocket.CONNECTING]: 'connecting',
+};

fix(websockets): don't push an event to the client that is  closing