imp(websockets): use verifyClient for connection verification
MODIFY
packages/@integration/src/ws.listener.ts
packages/@integration/src/ws.listener.ts
@@ -3,12 +3,13 @@ import { iif, throwError, of } from 'rxjs';
import { mergeMap } from 'rxjs/operators';
import { add$ } from './effects/calculator.ws-effects';
import { logger$ } from './middlewares/logger.ws-middleware';
+import { HttpStatus } from '@marblejs/core';
const connection$: WsConnectionEffect = req$ =>
req$.pipe(
mergeMap(req => iif(
() => req.headers.upgrade !== 'websocket',
- throwError(new WebSocketConnectionError('Unauthorized', 4000)),
+ throwError(new WebSocketConnectionError('Unauthorized', HttpStatus.UNAUTHORIZED)),
of(req),
)),
);

MODIFY
packages/websockets/src/effects/ws-effects.interface.ts
packages/websockets/src/effects/ws-effects.interface.ts
@@ -15,7 +15,7 @@ export interface WsErrorEffect<
export interface WsConnectionEffect<
T extends http.IncomingMessage = http.IncomingMessage
-> extends WsEffect<T, T, MarbleWebSocketClient> {}
+> extends WsEffect<T, T, undefined> {}
export interface WsOutputEffect<
T extends Event = Event

MODIFY
packages/websockets/src/listener/specs/websocket.listener.spec.ts
packages/websockets/src/listener/specs/websocket.listener.spec.ts
-import { EventError, createContext } from '@marblejs/core';
-import { throwError, fromEvent, forkJoin } from 'rxjs';
-import { tap, map, mergeMap, first, toArray, take } from 'rxjs/operators';
+import { EventError, createContext, HttpStatus } from '@marblejs/core';
+import { throwError, fromEvent, forkJoin, merge } from 'rxjs';
+import { tap, map, mergeMap, first, toArray, take, mergeMapTo } from 'rxjs/operators';
import { webSocketListener } from '../websocket.listener';
import { WsEffect, WsMiddlewareEffect, WsConnectionEffect } from '../../effects/ws-effects.interface';
import { WebSocketConnectionError } from '../../error/ws-error.model';
@@ -170,25 +170,55 @@ describe('WebSocket listener', () => {
});
});
+ test('passes connection', done => {
+ // given
+ const connection$: WsConnectionEffect = req$ => req$;
+ const webSocketServer = webSocketListener({ connection$ });
+ const targetClient1 = testBed.getClient(0);
+ const targetClient2 = testBed.getClient(1);
+ const server = testBed.getServer();
+ const context = createContext();
+
+ // when
+ webSocketServer({ server }).run(context);
+
+ // then
+ merge(
+ fromEvent(targetClient1, 'open'),
+ fromEvent(targetClient2, 'open'),
+ )
+ .pipe(take(2), toArray())
+ .subscribe(() => done());
+ });
+
test('triggers connection error', done => {
// given
- const error = new WebSocketConnectionError('Unauthorized', 4000);
- const connection$: WsConnectionEffect = req$ => req$.pipe(mergeMap(() => throwError(error)));
+ const error = new WebSocketConnectionError('Unauthorized', HttpStatus.UNAUTHORIZED);
+ const connection$: WsConnectionEffect = req$ => req$.pipe(mergeMapTo(throwError(error)));
const webSocketServer = webSocketListener({ connection$ });
- const targetClient = testBed.getClient(0);
+ const targetClient1 = testBed.getClient(0);
+ const targetClient2 = testBed.getClient(1);
const server = testBed.getServer();
const context = createContext();
// when
webSocketServer({ server }).run(context);
- targetClient.once('open', () => targetClient.send('test'));
// then
- targetClient.once('close', (status, message) => {
- expect(status).toEqual(error.status);
- expect(message).toEqual(error.message);
+ merge(
+ fromEvent(targetClient1, 'unexpected-response'),
+ fromEvent(targetClient2, 'unexpected-response'),
+ )
+ .pipe(take(2), toArray())
+ .subscribe(
+ (data) => {
+ expect(data[0][1].statusCode).toEqual(error.status);
+ expect(data[1][1].statusCode).toEqual(error.status);
+ expect(data[0][1].statusMessage).toEqual(error.message);
+ expect(data[1][1].statusMessage).toEqual(error.message);
done();
- });
+ },
+ );
});
});

MODIFY
packages/websockets/src/listener/websocket.listener.ts
packages/websockets/src/listener/websocket.listener.ts
import * as http from 'http';
import * as WebSocket from 'ws';
-import { Subject, of, Observable } from 'rxjs';
+import { Subject, Observable, of } from 'rxjs';
import { map } from 'rxjs/operators';
import {
combineEffects,
@@ -19,12 +19,12 @@ import { EventTransformer } from '../transformer/transformer.inteface';
import { handleResponse, handleBroadcastResponse } from '../response/ws-response.handler';
import { handleEffectsError } from '../error/ws-error.handler';
import { provideErrorEffect } from '../error/ws-error.provider';
+import { WebSocketConnectionError } from '../error/ws-error.model';
-type HandleIncomingMessage =
- (client: WS.MarbleWebSocketClient, contextProvider: ContextProvider) =>
- () => void;
+type HandleMessage =
+ (client: WS.MarbleWebSocketClient, contextProvider: ContextProvider) => void;
-type HandleIncomingConnection =
+type HandleConnection =
(server: WS.MarbleWebSocketServer, contextProvider: ContextProvider) =>
(client: WS.WebSocketClient, request: http.IncomingMessage) => void;
@@ -52,7 +52,7 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
const providedError$ = provideErrorEffect(error$, eventTransformer);
const providedTransformer: EventTransformer<any, any> = eventTransformer || jsonTransformer;
- const handleIncomingMessage: HandleIncomingMessage = (client, ask) => () => {
+ const handleMessage: HandleMessage = (client, ask) => {
const eventSubject$ = new Subject<any>();
const incomingEventSubject$ = new Subject<WS.WebSocketData>();
const defaultMetadata = createEffectMetadata({ ask });
@@ -92,29 +92,36 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
client.once('close', onClose);
};
- const handleIncomingConnection: HandleIncomingConnection = (server, ask) => (client, req) => {
- const request$ = of(req);
- const defaultMetadata = createEffectMetadata({ ask });
+ const handleConnection: HandleConnection = (server, ask) => (client) => {
const extendedClient = WSHelper.extendClientWith({
sendResponse: handleResponse(client, providedTransformer),
sendBroadcastResponse: handleBroadcastResponse(server, providedTransformer),
isAlive: true,
})(client);
- connection$(request$, extendedClient, defaultMetadata).subscribe(
- handleIncomingMessage(extendedClient, ask),
- WSHelper.handleClientValidationError(extendedClient),
- );
-
+ handleMessage(extendedClient, ask);
WSHelper.handleClientBrokenConnection(extendedClient).subscribe();
};
- return (serverOptions: WebSocket.ServerOptions = { noServer: true }): ContextReader => reader.map(ask => {
- const webSocketServer = WSHelper.createWebSocketServer(serverOptions);
+ const verifyClient = (ask: ContextProvider): WebSocket.VerifyClientCallbackAsync => (info, callback) => {
+ connection$(of(info.req), undefined, createEffectMetadata({ ask }))
+ .pipe(map(Boolean))
+ .subscribe(
+ isVerified => callback(isVerified),
+ (error: WebSocketConnectionError) => callback(false, error.status, error.message),
+ );
+ };
+
+ return (serverOptions: WebSocket.ServerOptions = {}): ContextReader => reader.map(ask => {
+ const webSocketServer = WSHelper.createWebSocketServer({
+ noServer: true,
+ verifyClient: verifyClient(ask),
+ ...serverOptions,
+ });
const sendBroadcastResponse = handleBroadcastResponse(webSocketServer, providedTransformer);
const extendedWebSocketServer = WSHelper.extendServerWith({ sendBroadcastResponse })(webSocketServer);
- extendedWebSocketServer.on('connection', handleIncomingConnection(extendedWebSocketServer, ask));
+ extendedWebSocketServer.on('connection', handleConnection(extendedWebSocketServer, ask));
WSHelper.handleServerBrokenConnections(extendedWebSocketServer).subscribe();
return extendedWebSocketServer;

imp(websockets): use verifyClient for connection verification