imp(@integration): moved websocket related code to separate folder
MODIFY
packages/@integration/package.json
packages/@integration/package.json
"main": "src/index.ts",
"scripts": {
"start:http": "ts-node --project tsconfig.json src/http/index.ts",
+ "start:websockets": "ts-node --project tsconfig.json src/websockets/http.server.ts",
"start:messaging:server": "ts-node --project tsconfig.json src/messaging/server.ts",
"start:messaging:client": "ts-node --project tsconfig.json src/messaging/client.ts",
"watch:dev": "nodemon --watch src --ext ts --ignore '*.spec.ts' --exec yarn start",

MODIFY
packages/@integration/src/http/effects/api.effects.ts
packages/@integration/src/http/effects/api.effects.ts
import { r, HttpError, HttpStatus, combineRoutes, use, switchToProtocol } from '@marblejs/core';
import { requestValidator$, t } from '@marblejs/middleware-io';
import { throwError } from 'rxjs';
-import { map, mergeMap, tap } from 'rxjs/operators';
+import { map, mergeMap } from 'rxjs/operators';
import { user$ } from './user.effects';
import { static$ } from './static.effects';
-import { WsServerToken } from '../tokens';
const rootValiadtor$ = requestValidator$({
params: t.type({
@@ -18,13 +17,10 @@ const rootValiadtor$ = requestValidator$({
const root$ = r.pipe(
r.matchPath('/'),
r.matchType('GET'),
- r.useEffect((req$, _, { ask }) => req$.pipe(
+ r.useEffect(req$ => req$.pipe(
use(rootValiadtor$),
map(req => req.params.version),
map(version => `API version: ${version}`),
- tap(message => ask(WsServerToken).map(server =>
- server.sendBroadcastResponse({ type: 'ROOT', payload: message })),
- ),
map(message => ({ body: message })),
)));

DELETE
packages/@integration/src/http/effects/calculator.ws-effects.ts
-import { use, matchEvent } from '@marblejs/core';
-import { WsEffect } from '@marblejs/websockets';
-import { t, eventValidator$ } from '@marblejs/middleware-io';
-import { buffer, map } from 'rxjs/operators';
-
-export const sum$: WsEffect = event$ =>
- event$.pipe(
- matchEvent('SUM')
- );
-
-export const add$: WsEffect = (event$, ...args) =>
- event$.pipe(
- matchEvent('ADD'),
- use(eventValidator$(t.number)),
- buffer(sum$(event$, ...args)),
- map(addEvents => addEvents.reduce((a, e) => e.payload + a, 0)),
- map(payload => ({ type: 'SUM_RESULT', payload })),
- );

MODIFY
packages/@integration/src/http/effects/static.effects.ts
packages/@integration/src/http/effects/static.effects.ts
@@ -7,8 +7,8 @@ import { streamFileTo } from '@marblejs/middleware-multipart/dist/multipart.util
import { readFile } from '@marblejs/core/dist/+internal';
import { map, mergeMap } from 'rxjs/operators';
-const STATIC_PATH = path.resolve(__dirname, '../../../../assets');
-const TMP_PATH = path.resolve(__dirname, '../../../../tmp');
+const STATIC_PATH = path.resolve(__dirname, '../../../../../assets');
+const TMP_PATH = path.resolve(__dirname, '../../../../../tmp');
const getFileValidator$ = requestValidator$({
params: t.type({ dir: t.string })

MODIFY
packages/@integration/src/http/index.ts
packages/@integration/src/http/index.ts
-import { createServer, matchEvent, ServerEvent, HttpServerEffect, bindTo } from '@marblejs/core';
-import { mapToServer } from '@marblejs/websockets';
+import { createServer, matchEvent, ServerEvent, HttpServerEffect } from '@marblejs/core';
import { merge } from 'rxjs';
import { tap, map } from 'rxjs/operators';
-import { WsServerToken } from './tokens';
import httpListener from './http.listener';
-import webSocketListener from './ws.listener';
-
-const upgrade$: HttpServerEffect = (event$, _, { ask }) =>
- event$.pipe(
- matchEvent(ServerEvent.upgrade),
- mapToServer({
- path: '/api/:version/ws',
- server: ask(WsServerToken),
- }),
- );
const listening$: HttpServerEffect = event$ =>
event$.pipe(
@@ -25,12 +13,8 @@ const listening$: HttpServerEffect = event$ =>
export const server = createServer({
port: 1337,
httpListener,
- dependencies: [
- bindTo(WsServerToken)(webSocketListener().run),
- ],
event$: (...args) => merge(
listening$(...args),
- upgrade$(...args),
),
});

MODIFY
packages/@integration/src/http/middlewares/logger.middleware.ts
packages/@integration/src/http/middlewares/logger.middleware.ts
@@ -3,7 +3,7 @@ import { createWriteStream } from 'fs';
import { join } from 'path';
const silent = process.env.NODE_ENV === 'test';
-const writePath = join(__dirname, '../../', 'access.log');
+const writePath = join(__dirname, '../../../', 'access.log');
const stream = createWriteStream(writePath, { flags: 'a' });
export const loggerDev$ = logger$({ silent });

DELETE
packages/@integration/src/http/middlewares/logger.ws-middleware.ts
-import { WsMiddlewareEffect } from '@marblejs/websockets';
-import { tap } from 'rxjs/operators';
-
-export const logger$: WsMiddlewareEffect = event$ =>
- event$.pipe(
- tap(e => console.log(`type: ${e.type}, payload: ${e.payload}`)),
- );

DELETE
packages/@integration/src/http/tokens.ts
-import { createContextToken } from '@marblejs/core';
-import { MarbleWebSocketServer } from '@marblejs/websockets';
-
-export const WsServerToken = createContextToken<MarbleWebSocketServer>();

ADD
packages/@integration/src/websockets/websockets.server.ts
+import {
+ WsEffect,
+ WsMiddlewareEffect,
+ WsConnectionEffect,
+ WebSocketConnectionError,
+ webSocketListener,
+} from '@marblejs/websockets';
+import { iif, throwError, of } from 'rxjs';
+import { tap, mergeMap, buffer, map } from 'rxjs/operators';
+import { matchEvent, use } from '@marblejs/core';
+import { eventValidator$, t } from '../../../middleware-io/dist';
+
+const sum$: WsEffect = event$ =>
+ event$.pipe(
+ matchEvent('SUM')
+ );
+
+const add$: WsEffect = (event$, ...args) =>
+ event$.pipe(
+ matchEvent('ADD'),
+ use(eventValidator$(t.number)),
+ buffer(sum$(event$, ...args)),
+ map(addEvents => addEvents.reduce((a, e) => e.payload + a, 0)),
+ 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(
+ () => req.headers.upgrade !== 'websocket',
+ throwError(new WebSocketConnectionError('Unauthorized', 4000)),
+ of(req),
+ )),
+ );
+
+export const websocketsServer = webSocketListener({
+ middlewares: [logger$],
+ effects: [add$],
+ connection$,
+})();

imp(@integration): moved websocket related code to separate folder