refactor(websockets): server integration tests (removed testbed)
MODIFY
packages/@integration/src/websockets/http.server.ts
packages/@integration/src/websockets/http.server.ts
@@ -8,12 +8,12 @@ import {
httpListener,
HttpServerEffect,
useContext,
+ combineEffects,
} from '@marblejs/core';
import { mapToServer, WebSocketServerConnection } from '@marblejs/websockets';
import { logger$ } from '@marblejs/middleware-logger';
import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
import { IO } from 'fp-ts/lib/IO';
-import { merge } from 'rxjs';
import { tap, map } from 'rxjs/operators';
import { webSocketServer } from './websockets.server';
@@ -22,8 +22,8 @@ export const WebSocketServerToken = createContextToken<WebSocketServerConnection
const root$ = r.pipe(
r.matchPath('/'),
r.matchType('GET'),
- r.useEffect((req$, { ask }) => {
- const webSocketServer = useContext(WebSocketServerToken)(ask);
+ r.useEffect((req$, ctx) => {
+ const webSocketServer = useContext(WebSocketServerToken)(ctx.ask);
return req$.pipe(
tap(() => webSocketServer.sendBroadcastResponse({ type: 'ROOT', payload: 'Hello' })),
@@ -31,12 +31,12 @@ const root$ = r.pipe(
);
}));
-const upgrade$: HttpServerEffect = (event$, { ask }) =>
+const upgrade$: HttpServerEffect = (event$, ctx) =>
event$.pipe(
matchEvent(ServerEvent.upgrade),
mapToServer({
path: '/api/:version/ws',
- server: ask(WebSocketServerToken),
+ server: ctx.ask(WebSocketServerToken),
}),
);
@@ -47,13 +47,12 @@ export const server = createServer({
effects: [root$],
}),
dependencies: [
- bindEagerlyTo(WebSocketServerToken)(async () => {
- const app = await webSocketServer;
- return app();
- }),
+ bindEagerlyTo(WebSocketServerToken)(async () =>
+ (await webSocketServer)()
+ ),
],
- event$: (...args) => merge(
- upgrade$(...args),
+ event$: combineEffects(
+ upgrade$,
),
});

MODIFY
packages/middleware-io/test/io-ws.integration.spec.ts
packages/middleware-io/test/io-ws.integration.spec.ts
-import { createWebSocketsTestBed } from '@marblejs/websockets/dist/+internal';
-import { createWebSocketServer } from '@marblejs/websockets';
+import { bootstrapHttpServer, bootstrapWebSocketClient, bootstrapWebSocketServer } from '@marblejs/websockets/dist/+internal';
import { listener } from './io-ws.integration';
describe('@marblejs/middleware-io - WebSocket integration', () => {
- const testBed = createWebSocketsTestBed();
-
- beforeEach(testBed.bootstrap);
- afterEach(testBed.teardown);
test('[POST_USER] sends user object', async done => {
// given
const user = { id: 'id', age: 100, };
- const event = JSON.stringify({ type: 'POST_USER', payload: user });
- const server = testBed.getServer();
- const targetClient = testBed.getClient();
+ const event = { type: 'POST_USER', payload: user };
// when
- const app = await createWebSocketServer({ options: { server }, listener });
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer}, listener);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
- targetClient.once('open', () => targetClient.send(event));
+ webSocketClient.send(JSON.stringify(event));
// then
- targetClient.once('message', message => {
- expect(message).toEqual(event);
- webSocketServer.close(done);
+ webSocketClient.once('message', message => {
+ expect(JSON.parse(message)).toEqual(event);
+ webSocketClient.close();
+ webSocketServer.close();
+ httpServer.close();
+ done();
});
});
test('[POST_USER] throws an error if incoming object is invalid', async done => {
// given
- const server = testBed.getServer();
- const targetClient = testBed.getClient();
const user = { id: 'id', age: '100', };
- const event = JSON.stringify({ type: 'POST_USER', payload: user });
+ const event = { type: 'POST_USER', payload: user };
const expectedError = {
type: 'POST_USER',
error: {
@@ -44,16 +39,19 @@ describe('@marblejs/middleware-io - WebSocket integration', () => {
};
// when
- const app = await createWebSocketServer({ options: { server }, listener });
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer}, listener);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
- targetClient.once('open', () => targetClient.send(event));
+ webSocketClient.send(JSON.stringify(event));
// then
- targetClient.once('message', message => {
- const parsedMessage = JSON.parse(message);
- expect(parsedMessage).toEqual(expectedError);
- webSocketServer.close(done);
+ webSocketClient.once('message', message => {
+ expect(JSON.parse(message)).toEqual(expectedError);
+ webSocketClient.close();
+ webSocketServer.close();
+ httpServer.close();
+ done();
});
});
});

MODIFY
packages/websockets/src/+internal/websocket.test.util.ts
packages/websockets/src/+internal/websocket.test.util.ts
import * as http from 'http';
+import * as net from 'net';
import * as WebSocket from 'ws';
-import { WebSocketClientConnection, WebSocketServerConnection } from '../server/websocket.server.interface';
-
-export const TEST_CONFIG = {
- PORT: 1337,
- HOST: '127.0.0.1',
-};
-
-const createServer = (host: string, port: number) => (cb: () => void) =>
- http.createServer().listen(port, host, cb);
-
-const createWebSocketClient = (host: string, port: number) => () =>
- new WebSocket(`ws://${host}:${port}`);
-
-export const createWebSocketsTestBed = (clientsCount = 1, host = TEST_CONFIG.HOST, port = TEST_CONFIG.PORT) => {
- let httpServer: http.Server;
- let webSocketClients: WebSocket[] = [];
-
- const bootstrap = (cb: () => void) => {
- httpServer = createServer(host, port)(() => {
- webSocketClients = Array.from(
- { length: clientsCount },
- createWebSocketClient(host, port)
- );
- cb();
+import { ServerConfig } from '@marblejs/core';
+import { WebSocketClientConnection, WebSocketServerConnection, WebSocketServerConfig } from '../server/websocket.server.interface';
+import { createWebSocketServer } from '../server/websocket.server';
+import { WsServerEffect } from '../effects/websocket.effects.interface';
+
+export const bootstrapHttpServer = async (host?: string, port?: number): Promise<http.Server> =>
+ new Promise(resolve => {
+ const httpServer = http.createServer();
+ httpServer.listen(port, host, () => resolve(httpServer));
});
- };
-
- const teardown = (cb: () => void) => {
- webSocketClients.forEach(client => client.readyState === WebSocket.OPEN && client.close());
- webSocketClients = [];
- httpServer.close(cb);
- };
- const getClient = (index = 0) =>
- webSocketClients[index];
-
- const getServer = () =>
- httpServer;
-
- return {
- bootstrap,
- teardown,
- getClient,
- getServer,
- };
+export const bootstrapWebSocketClient = (httpServer: http.Server): Promise<WebSocket> => {
+ const serverAddressInfo = httpServer.address() as net.AddressInfo;
+ const host = serverAddressInfo.address === '::' ? '127.0.0.1' : serverAddressInfo.address;
+ const port = serverAddressInfo.port;
+ const client = new WebSocket(`ws://${host}:${port}`);
+ return new Promise(resolve => client.once('open', () => resolve(client)));
};
-export type WebSocketsTestBed = typeof createWebSocketsTestBed;
+export const bootstrapWebSocketServer = async (
+ options: WebSocketServerConfig['options'],
+ listener: ServerConfig<any, any>['listener'],
+ event$?: WsServerEffect,
+) => (await createWebSocketServer({ options, listener, event$ }))();
export const createWebSocketClientMock = (): WebSocketClientConnection => {
class WebSocketClientMock extends WebSocket.EventEmitter {

ADD
packages/websockets/src/server/specs/websocket.server.event.spec.ts
+import { tap } from 'rxjs/operators';
+import { matchEvent } from '@marblejs/core';
+import { ServerEvent, ServerEventType } from '../websocket.server.event';
+import { webSocketListener } from '../websocket.server.listener';
+import { bootstrapHttpServer, bootstrapWebSocketClient, bootstrapWebSocketServer } from '../../+internal';
+import { WsServerEffect } from '../../effects/websocket.effects.interface';
+
+describe('WebSocket events', () => {
+
+ test(`emits "${ServerEventType.LISTENING}" when no server passed`, async done => {
+ const listening$: WsServerEffect = event$ =>
+ event$.pipe(
+ matchEvent(ServerEvent.listening),
+ tap(event => {
+ expect(event.type).toEqual(ServerEventType.LISTENING);
+ expect(event.payload.host).toEqual(expect.any(String));
+ expect(event.payload.port).toEqual(expect.any(Number));
+ done();
+ }),
+ );
+
+ const webSocketServer = await bootstrapWebSocketServer({ port: 8060 }, webSocketListener(), listening$);
+ webSocketServer.close();
+ });
+
+ test(`emits "${ServerEventType.LISTENING}" when server passed`, async done => {
+ const listening$: WsServerEffect = event$ =>
+ event$.pipe(
+ matchEvent(ServerEvent.listening),
+ tap(event => {
+ expect(event.type).toEqual(ServerEventType.LISTENING);
+ expect(event.payload.host).toEqual(expect.any(String));
+ expect(event.payload.port).toEqual(expect.any(Number));
+ done();
+ }),
+ );
+
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, webSocketListener(), listening$);
+ webSocketServer.close();
+ httpServer.close();
+ });
+
+ test(`emits "${ServerEventType.ERROR}"`, async done => {
+ const error = new Error('test');
+ const error$: WsServerEffect = event$ =>
+ event$.pipe(
+ matchEvent(ServerEvent.error),
+ tap(event => {
+ expect(event.type).toEqual(ServerEventType.ERROR);
+ expect(event.payload.error).toBeDefined();
+ done();
+ }),
+ );
+
+ const webSocketServer = await bootstrapWebSocketServer({ port: 8060 }, webSocketListener(), error$);
+ webSocketServer.emit('error', error);
+ webSocketServer.close();
+ });
+
+ test(`emits "${ServerEventType.HEADERS}"`, async () => {
+ const headers$: WsServerEffect = event$ =>
+ event$.pipe(
+ matchEvent(ServerEvent.headers),
+ tap(event => {
+ expect(event.type).toEqual(ServerEventType.HEADERS);
+ expect(event.payload.headers).toBeDefined();
+ expect(event.payload.req).toBeDefined();
+ }),
+ );
+
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, webSocketListener(), headers$);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
+
+ webSocketServer.close();
+ webSocketClient.close();
+ httpServer.close();
+ });
+
+ test(`emits "${ServerEventType.CONNECTION}"`, async () => {
+ const connection$: WsServerEffect = event$ =>
+ event$.pipe(
+ matchEvent(ServerEvent.connection),
+ tap(event => {
+ expect(event.type).toEqual(ServerEventType.CONNECTION);
+ expect(event.payload.client).toBeDefined();
+ expect(event.payload.req).toBeDefined();
+ }),
+ );
+
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, webSocketListener(), connection$);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
+
+ webSocketServer.close();
+ webSocketClient.close();
+ httpServer.close();
+ });
+
+ test(`emits "${ServerEventType.CLOSE_CLIENT}"`, async done => {
+ const connection$: WsServerEffect = event$ =>
+ event$.pipe(
+ matchEvent(ServerEvent.closeClient),
+ tap(event => {
+ expect(event.type).toEqual(ServerEventType.CLOSE_CLIENT);
+ expect(event.payload.client).toBeDefined();
+
+ // teardown
+ webSocketServer.close();
+ httpServer.close();
+ done();
+ }),
+ );
+
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, webSocketListener(), connection$);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
+ webSocketClient.close();
+ });
+});

MODIFY
packages/websockets/src/server/specs/websocket.server.spec.ts
packages/websockets/src/server/specs/websocket.server.spec.ts
-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 { Event } from '@marblejs/core';
+import { throwError, fromEvent, forkJoin } from 'rxjs';
+import { tap, map, mergeMap, first, toArray, take, mapTo } from 'rxjs/operators';
import { webSocketListener } from '../websocket.server.listener';
-import { WsEffect, WsMiddlewareEffect, WsConnectionEffect } from '../../effects/websocket.effects.interface';
-import { WebSocketConnectionError } from '../../error/websocket.error.model';
+import { WsEffect, WsMiddlewareEffect } from '../../effects/websocket.effects.interface';
import { EventTransformer } from '../../transformer/websocket.transformer.interface';
-import { createWebSocketsTestBed } from '../../+internal';
-import { createWebSocketServer } from '../websocket.server';
+import { bootstrapWebSocketClient, bootstrapHttpServer, bootstrapWebSocketServer } from '../../+internal';
describe('WebSocket server', () => {
describe('JSON transformer', () => {
- const testBed = createWebSocketsTestBed(2);
-
- beforeEach(testBed.bootstrap);
- afterEach(testBed.teardown);
-
test('echoes back', async done => {
// given
- const targetClient = testBed.getClient(0);
- const server = testBed.getServer();
const echo$: WsEffect = event$ => event$;
- const event = JSON.stringify({ type: 'EVENT', payload: 'test' });
+ const event = { type: 'EVENT', payload: 'test' };
const listener = webSocketListener({ effects: [echo$] });
// when
- const app = await createWebSocketServer({ options: { server }, listener });
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, listener);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
- targetClient.once('open', () => targetClient.send(event));
+ webSocketClient.send(JSON.stringify(event));
// then
- targetClient.once('message', message => {
- expect(message).toEqual(event);
- webSocketServer.close(done);
+ webSocketClient.once('message', message => {
+ expect(JSON.parse(message)).toEqual(event);
+ webSocketServer.close();
+ webSocketClient.close();
+ httpServer.close();
+ done();
});
});
test('echoes back to all clients', async done => {
// given
- const echo$: WsEffect = (event$, { client }) => event$.pipe(
- mergeMap(event => client.sendBroadcastResponse(event).pipe(mapTo(event))),
+ const echo$: WsEffect = (event$, ctx) => event$.pipe(
+ mergeMap(event => ctx.client.sendBroadcastResponse({ type: event.type, payload: event.payload }).pipe(mapTo(event))),
);
- const event = { type: 'EVENT', payload: 'test' } as Event;
+ const event = { type: 'EVENT', payload: 'test' };
const listener = webSocketListener({ effects: [echo$] });
- const server = testBed.getServer();
- const targetClient = testBed.getClient(0);
// when
- const app = await createWebSocketServer({ options: { server }, listener });
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, listener);
+ const webSocketclient1 = await bootstrapWebSocketClient(httpServer);
+ const webSocketclient2 = await bootstrapWebSocketClient(httpServer);
- targetClient.on('open', () => targetClient.send(JSON.stringify(event)));
+ webSocketclient1.send(JSON.stringify(event));
// then
- const client1$ = fromEvent(testBed.getClient(0), 'message').pipe(first());
- const client2$ = fromEvent(testBed.getClient(1), 'message').pipe(first());
+ const client1$ = fromEvent(webSocketclient1, 'message').pipe(first());
+ const client2$ = fromEvent(webSocketclient2, 'message').pipe(first());
forkJoin(client1$, client2$).subscribe(([ message1, message2 ]: [any, any]) => {
- expect(JSON.parse(message1.data)).toEqual(expect.objectContaining(event));
- expect(JSON.parse(message2.data)).toEqual(expect.objectContaining(event));
- webSocketServer.close(done);
+ expect(JSON.parse(message1.data)).toEqual(event);
+ expect(JSON.parse(message2.data)).toEqual(event);
+ webSocketclient1.close();
+ webSocketclient2.close();
+ webSocketServer.close();
+ httpServer.close();
+ done();
});
});
test('echoes back on upgraded http server', async done => {
// given
const echo$: WsEffect = event$ => event$;
- const event = JSON.stringify({ type: 'EVENT', payload: 'test' });
- const server = testBed.getServer();
+ const event = { type: 'EVENT', payload: 'test' };
const listener = webSocketListener({ effects: [echo$] });
- const targetClient = testBed.getClient(0);
// when
- const app = await createWebSocketServer({ listener });
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer(undefined, listener);
- server.on('upgrade', (request, socket, head) => {
+ httpServer.on('upgrade', (request, socket, head) => {
webSocketServer.handleUpgrade(request, socket, head, ws => {
webSocketServer.emit('connection', ws, request);
});
});
- targetClient.once('open', () => targetClient.send(event));
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
+
+ webSocketClient.send(JSON.stringify(event));
// then
- targetClient.once('message', message => {
- expect(message).toEqual(event);
- webSocketServer.close(done);
+ webSocketClient.once('message', message => {
+ expect(JSON.parse(message)).toEqual(event);
+ webSocketServer.close();
+ webSocketClient.close();
+ httpServer.close();
+ done();
});
});
test('passes through middlewares', async done => {
// given
- const incomingEvent = JSON.stringify({ type: 'EVENT', payload: 0 });
- const outgoingEvent = JSON.stringify({ type: 'EVENT', payload: 3 });
+ const incomingEvent = { type: 'EVENT', payload: 0 };
+ const outgoingEvent = { type: 'EVENT', payload: 3 };
const e$: WsEffect = event$ => event$;
const m$: WsMiddlewareEffect = event$ => event$.pipe(
tap(event => event.payload !== undefined && (event.payload as number)++)
);
- const targetClient = testBed.getClient(0);
- const server = testBed.getServer();
const listener = webSocketListener({
effects: [e$],
middlewares: [m$, m$, m$],
});
// when
- const app = await createWebSocketServer({ options: { server }, listener })
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, listener);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
- targetClient.once('open', () => targetClient.send(incomingEvent));
+ webSocketClient.send(JSON.stringify(incomingEvent));
// then
- targetClient.once('message', message => {
- expect(message).toEqual(outgoingEvent);
- webSocketServer.close(done);
+ webSocketClient.once('message', message => {
+ expect(JSON.parse(message)).toEqual(outgoingEvent);
+ webSocketClient.close();
+ webSocketServer.close();
+ httpServer.close();
+ done();
});
});
test('passes error (thrown by invalid JSON object) through stream multiple times', async done => {
// given
const incomingEvent = '{ some: wrong JSON object }';
- const outgoingEvent = JSON.stringify({
+ const outgoingEvent = {
type: 'UNHANDLED_ERROR',
error: { name: 'SyntaxError', message: 'Unexpected token s in JSON at position 2' },
- });
- const targetClient = testBed.getClient(0);
- const server = testBed.getServer();
- const listener = webSocketListener();
+ };
// when
- const app = await createWebSocketServer({ options: { server }, listener });
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, webSocketListener());
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
- targetClient.once('open', () => {
- targetClient.send(incomingEvent);
- targetClient.send(incomingEvent);
- });
+ webSocketClient.send(incomingEvent);
+ webSocketClient.send(incomingEvent);
// then
- fromEvent(targetClient, 'message')
+ fromEvent(webSocketClient, 'message')
.pipe(take(2), toArray())
.subscribe((messages: any[]) => {
- expect(messages[0].data).toEqual(outgoingEvent);
- expect(messages[1].data).toEqual(outgoingEvent);
- webSocketServer.close(done);
+ expect(JSON.parse(messages[0].data)).toEqual(outgoingEvent);
+ expect(JSON.parse(messages[1].data)).toEqual(outgoingEvent);
+ webSocketClient.close();
+ webSocketServer.close();
+ httpServer.close();
+ done();
});
});
test('passes error (thrown by effect) through stream multiple times', async done => {
// given
- const incomingEvent = JSON.stringify({ type: 'EVENT' });
- const outgoingEvent = JSON.stringify({ type: 'UNHANDLED_ERROR', error: { name: 'Error', message: 'test_message' } });
+ const incomingEvent = { type: 'EVENT' };
+ const outgoingEvent = { type: 'UNHANDLED_ERROR', error: { name: 'Error', message: 'test_message' } };
const effect$: WsEffect = event$ =>
- event$.pipe(
- mergeMap(() => throwError(new Error('test_message'))),
- );
+ event$.pipe(mergeMap(() => throwError(new Error('test_message'))));
- const targetClient = testBed.getClient(0);
- const server = testBed.getServer();
const listener = webSocketListener({ effects: [effect$] });
// when
- const app = await createWebSocketServer({ options: { server }, listener });
- const webSocketServer = await app();
+ const httpSever = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpSever }, listener);
+ const webSocketClient = await bootstrapWebSocketClient(httpSever);
- targetClient.once('open', () => {
- targetClient.send(incomingEvent);
- targetClient.send(incomingEvent);
- });
+ webSocketClient.send(JSON.stringify(incomingEvent));
+ webSocketClient.send(JSON.stringify(incomingEvent));
// then
- fromEvent(targetClient, 'message')
+ fromEvent(webSocketClient, 'message')
.pipe(take(2), toArray())
.subscribe((messages: any[]) => {
- expect(messages[0].data).toEqual(outgoingEvent);
- expect(messages[1].data).toEqual(outgoingEvent);
- webSocketServer.close(done);
+ expect(JSON.parse(messages[0].data)).toEqual(outgoingEvent);
+ expect(JSON.parse(messages[1].data)).toEqual(outgoingEvent);
+ webSocketClient.close();
+ webSocketServer.close();
+ httpSever.close();
+ done();
});
});
-
- test('passes connection', async done => {
- // given
- const connection$: WsConnectionEffect = req$ => req$;
- const listener = webSocketListener();
- const targetClient1 = testBed.getClient(0);
- const targetClient2 = testBed.getClient(1);
- const server = testBed.getServer();
-
- // when
- const app = await createWebSocketServer({ options: { server }, connection$, listener });
- const webSocketServer = await app();
-
- // then
- merge(
- fromEvent(targetClient1, 'open'),
- fromEvent(targetClient2, 'open'),
- )
- .pipe(take(2), toArray())
- .subscribe(() => webSocketServer.close(done));
- });
-
- test('triggers connection error', async done => {
- // given
- const error = new WebSocketConnectionError('Unauthorized', HttpStatus.UNAUTHORIZED);
- const connection$: WsConnectionEffect = req$ => req$.pipe(mergeMapTo(throwError(error)));
- const listener = webSocketListener();
- const targetClient1 = testBed.getClient(0);
- const targetClient2 = testBed.getClient(1);
- const server = testBed.getServer();
-
- // when
- const app = await createWebSocketServer({ options: { server }, connection$, listener });
- const webSocketServer = await app();
-
- // then
- merge(
- fromEvent(targetClient1, 'unexpected-response'),
- fromEvent(targetClient2, 'unexpected-response'),
- )
- .pipe(take(2), toArray())
- .subscribe(
- (data: any) => {
- 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);
- webSocketServer.close(done);
- },
- );
- });
});
describe('binary transformer', () => {
- const testBed = createWebSocketsTestBed();
-
- beforeEach(testBed.bootstrap);
- afterEach(testBed.teardown);
-
test('operates over binary events', async done => {
// given
- const targetClient = testBed.getClient();
const message = 'hello world';
const eventTransformer: EventTransformer<Buffer> = {
@@ -250,25 +194,24 @@ describe('WebSocket server', () => {
};
const effect$: WsEffect<Event<Buffer>, Event<Buffer>> = event$ =>
- event$.pipe(
- map(event => ({ ...event, payload: Buffer.from(message) })),
- );
+ event$.pipe(map(event => ({ ...event, payload: Buffer.from(message) })));
- const server = testBed.getServer();
const listener = webSocketListener({ effects: [effect$], eventTransformer });
// when
- const app = await createWebSocketServer({ options: { server }, listener });
- const webSocketServer = await app();
+ const httpServer = await bootstrapHttpServer();
+ const webSocketServer = await bootstrapWebSocketServer({ server: httpServer }, listener);
+ const webSocketClient = await bootstrapWebSocketClient(httpServer);
- targetClient.once('open', () => {
- targetClient.send(Buffer.from(message));
- });
+ webSocketClient.send(Buffer.from(message));
// then
- targetClient.once('message', (incomingMessage: Buffer) => {
+ webSocketClient.once('message', (incomingMessage: Buffer) => {
expect(incomingMessage.toString('utf8')).toEqual(message);
- webSocketServer.close(done);
+ webSocketClient.close();
+ webSocketServer.close();
+ httpServer.close();
+ done();
});
});
});

refactor(websockets): server integration tests (removed testbed)