imp(core, websockets): refactored mapToServer.operator + injector improvements
MODIFY
packages/@integration/src/effects/api.effects.ts
packages/@integration/src/effects/api.effects.ts
@@ -18,7 +18,7 @@ const rootValiadtor$ = requestValidator$({
const root$ = EffectFactory
.matchPath('/')
.matchType('GET')
- .use((req$, _, inject) => req$.pipe(
+ .use((req$, _, { inject }) => req$.pipe(
use(rootValiadtor$),
map(req => req.params.version),
map(version => `API version: ${version}`),

MODIFY
packages/@integration/src/index.ts
packages/@integration/src/index.ts
@@ -6,13 +6,13 @@ import { httpServer } from './http.listener';
import { webSocketServer } from './ws.listener';
import { WebSocketsToken } from './tokens';
-const upgrade$: ServerEffect = (event$, _, injector) =>
+const upgrade$: ServerEffect = (event$, _, { inject }) =>
event$.pipe(
matchEvent(ServerEvent.upgrade),
mapToServer({
path: '/api/:version/ws',
- server: WebSocketsToken,
- })(injector),
+ server: inject(WebSocketsToken),
+ }),
);
const listen$: ServerEffect = event$ =>

MODIFY
packages/core/src/effects/effects.interface.ts
packages/core/src/effects/effects.interface.ts
@@ -2,9 +2,15 @@ import * as http from 'http';
import { Observable } from 'rxjs';
import { HttpRequest, HttpResponse, HttpStatus, HttpHeaders } from '../http.interface';
import { HttpError } from '../error/error.model';
-import { InjectionGetter } from '../server/server.injector';
+import { InjectorGetter } from '../server/server.injector';
import { Event } from '../event/event.interface';
+export interface EffectMetadata<T extends Error> {
+ error?: T;
+ inject: InjectorGetter;
+ [key: string]: any;
+}
+
export interface EffectHttpResponse<T = any> {
status?: HttpStatus;
headers?: HttpHeaders;
@@ -26,7 +32,7 @@ export interface Effect<
T = HttpRequest,
U = EffectHttpResponse,
V = HttpResponse,
- W = InjectionGetter,
+ W extends Error = Error,
> {
- (input$: Observable<T>, client: V, meta: W): Observable<U>;
+ (input$: Observable<T>, client: V, meta: EffectMetadata<W>): Observable<U>;
}

MODIFY
packages/core/src/error/error.effect.ts
packages/core/src/error/error.effect.ts
-import { map } from 'rxjs/operators';
+import { map, mapTo } from 'rxjs/operators';
import { ErrorEffect } from '../effects/effects.interface';
import { HttpStatus } from '../http.interface';
import { HttpError, isHttpError } from './error.model';
@@ -18,9 +18,10 @@ const errorFactory = (status: HttpStatus, error: Error) =>
? { error: { status, message: error.message, data: error.data, context: error.context } }
: { error: { status, message: error.message } };
-export const defaultError$: ErrorEffect<HttpError> = (req$, _, error = defaultHttpError) => req$
+export const defaultError$: ErrorEffect<HttpError> = (req$, _, meta) => req$
.pipe(
- map(() => {
+ mapTo(meta.error || defaultHttpError),
+ map(error => {
const status = getStatusCode(error);
const body = errorFactory(status, error);
return { status, body };

MODIFY
packages/core/src/error/specs/error.effect.spec.ts
packages/core/src/error/specs/error.effect.spec.ts
@@ -7,7 +7,7 @@ describe('defaultError$', () => {
const client = createHttpResponse();
test('maps HttpError', () => {
- const incomingError = new HttpError('test-message', 400);
+ const error = new HttpError('test-message', 400);
const outgoingResponse = {
status: 400,
body: { error: {
@@ -19,11 +19,11 @@ describe('defaultError$', () => {
Marbles.assertEffect(defaultError$, [
['-a-', { a: incomingRequest }],
['-a-', { a: outgoingResponse }],
- ], { client, meta: incomingError });
+ ], { client, meta: { error } });
});
test('maps other errors', () => {
- const incomingError = new Error('test-message');
+ const error = new Error('test-message');
const outgoingResponse = {
status: 500,
body: { error: {
@@ -35,7 +35,7 @@ describe('defaultError$', () => {
Marbles.assertEffect(defaultError$, [
['-a-', { a: incomingRequest }],
['-a-', { a: outgoingResponse }],
- ], { client, meta: incomingError });
+ ], { client, meta: { error } });
});
test('maps to "Internal server error" if "error" is not provided', () => {
@@ -50,6 +50,6 @@ describe('defaultError$', () => {
Marbles.assertEffect(defaultError$, [
['-a-', { a: incomingRequest }],
['-a-', { a: outgoingResponse }],
- ], { client });
+ ], { client, meta: {} });
});
});

MODIFY
packages/core/src/listener/http.listener.ts
packages/core/src/listener/http.listener.ts
@@ -27,18 +27,19 @@ export const httpListener = ({
const routing = factorizeRouting(effects);
const injector = createStaticInjectionContainer();
const defaultResponse = { status: HttpStatus.NOT_FOUND } as EffectHttpResponse;
+ const effectDefaultMeta = { inject: injector.get };
const effect$ = requestSubject$.pipe(
mergeMap(({ req, res }) => {
res.send = handleResponse(res)(req);
- return combinedMiddlewares(of(req), res, injector.get).pipe(
+ return combinedMiddlewares(of(req), res, effectDefaultMeta).pipe(
takeWhile(() => !res.finished),
switchMap(resolveRouting(routing, injector.get)(res)),
defaultIfEmpty(defaultResponse),
tap(res.send),
catchError(error =>
- error$(of(req), res, error).pipe(
+ error$(of(req), res, { ...effectDefaultMeta, error }).pipe(
tap(res.send),
),
),

MODIFY
packages/core/src/router/router.resolver.ts
packages/core/src/router/router.resolver.ts
@@ -4,7 +4,7 @@ import { HttpMethod, HttpRequest, HttpResponse } from '../http.interface';
import { EffectHttpResponse } from '../effects/effects.interface';
import { RouteMatched, Routing, RoutingItem } from './router.interface';
import { queryParamsFactory } from './router.query.factory';
-import { InjectionGetter } from '../server/server.injector';
+import { InjectorGetter } from '../server/server.injector';
export { RoutingItem };
export const findRoute = (
@@ -37,7 +37,7 @@ export const findRoute = (
};
export const resolveRouting =
- (routing: Routing, inject: InjectionGetter) =>
+ (routing: Routing, inject: InjectorGetter) =>
(res: HttpResponse) =>
(req: HttpRequest): Observable<EffectHttpResponse> => {
if (res.finished) { return EMPTY; }
@@ -54,9 +54,9 @@ export const resolveRouting =
const middleware = routeMatched.middleware;
return middleware
- ? middleware(of(req), res, inject).pipe(
+ ? middleware(of(req), res, { inject }).pipe(
takeWhile(() => !res.finished),
- mergeMap(req => routeMatched.effect(of(req), res, inject))
+ mergeMap(req => routeMatched.effect(of(req), res, { inject }))
)
- : routeMatched.effect(of(req), res, inject);
+ : routeMatched.effect(of(req), res, { inject });
};

MODIFY
packages/core/src/server/server.factory.ts
packages/core/src/server/server.factory.ts
@@ -27,7 +27,7 @@ export const createServer = (config: CreateServerConfig) => {
}
if (event$) {
- event$(serverEvent$, server, injector.get).subscribe();
+ event$(serverEvent$, server, { inject: injector.get }).subscribe();
}
return {

MODIFY
packages/core/src/server/server.injector.ts
packages/core/src/server/server.injector.ts
import * as http from 'http';
export type Injectable = (httpServer: http.Server) => any;
-export type Injector = typeof createStaticInjectionContainer;
export type InjectionDependencies = { token: InjectionToken, factory: Injectable }[];
export type InjectionToken<T = any> = new() => Token<T>;
-export interface InjectionGetter {
- <T>(key: InjectionToken<T>): T;
+export interface InjectorGetter {
+ <T>(token: InjectionToken<T>): T;
}
export class Token<T = any> {
surrogate!: T;
}
-export const createInjectionToken = <T>() =>
- class InlineToken extends Token<T> {};
-
-export const createStaticInjectionContainer = () => {
- const dependencies = new Map<InjectionToken, any>();
+export class StaticInjector {
+ private dependencies = new Map<InjectionToken, any>();
- const register = <T>(token: InjectionToken<T>, factory: Injectable) => (httpServer: http.Server) =>
- dependencies.set(token, factory(httpServer));
+ register = <T>(token: InjectionToken<T>, factory: Injectable) => (httpServer: http.Server) =>
+ this.dependencies.set(token, factory(httpServer))
- const deregister = <T>(token: InjectionToken<T>) =>
- dependencies.delete(token);
+ deregister = <T>(token: InjectionToken<T>) =>
+ this.dependencies.delete(token)
- const get = <T>(token: InjectionToken<T>): T =>
- dependencies.get(token);
+ get = <T>(token: InjectionToken<T>): T =>
+ this.dependencies.get(token)
- const registerAll = (dependenciesToRegister: InjectionDependencies) => (httpServer: http.Server) =>
+ registerAll = (dependenciesToRegister: InjectionDependencies) => (httpServer: http.Server) =>
dependenciesToRegister.forEach(({ token, factory }) =>
- register(token, factory)(httpServer)
- );
-
- const deregisterAll = () =>
- dependencies.clear();
-
- return {
- get,
- register,
- registerAll,
- deregister,
- deregisterAll,
- };
-};
+ this.register(token, factory)(httpServer)
+ )
+
+ deregisterAll = () =>
+ this.dependencies.clear()
+}
+
+export const createInjectionToken = <T>() =>
+ class InlineToken extends Token<T> {};
+
+export const createStaticInjectionContainer = () =>
+ new StaticInjector();
export const bind = <T>(token: InjectionToken<T>) => ({
to: (factory: Injectable) => ({ token, factory })

MODIFY
packages/middleware-body/src/index.spec.ts
packages/middleware-body/src/index.spec.ts
@@ -7,6 +7,7 @@ const MockReq = require('mock-req');
describe('BodyParser middleware', () => {
const injector = createStaticInjectionContainer();
+ const effectMeta = { inject: injector.get };
beforeEach(() => {
spyOn(console, 'log').and.stub();
@@ -31,7 +32,7 @@ describe('BodyParser middleware', () => {
});
const req$ = of(request as HttpRequest);
const res = {} as HttpResponse;
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, effectMeta);
http$.subscribe(data => {
expect(data.body).toEqual({ test: 'test' });
@@ -49,7 +50,7 @@ describe('BodyParser middleware', () => {
});
const req$ = of(request as HttpRequest);
const res = {} as HttpResponse;
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, effectMeta);
http$.subscribe(data => {
expect(data.body).toEqual({
@@ -72,7 +73,7 @@ describe('BodyParser middleware', () => {
});
const req$ = of(request as HttpRequest);
const res = {} as HttpResponse;
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, effectMeta);
http$.subscribe(
() => {
@@ -97,7 +98,7 @@ describe('BodyParser middleware', () => {
});
const req$ = of(request as HttpRequest);
const res = {} as HttpResponse;
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, effectMeta);
http$.subscribe(data => {
expect(data.body).toEqual('test');
@@ -115,7 +116,7 @@ describe('BodyParser middleware', () => {
});
const req$ = of(request as HttpRequest);
const res = {} as HttpResponse;
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, effectMeta);
http$.subscribe(
() => {

MODIFY
packages/middleware-jwt/src/spec/jwt.middleware.spec.ts
packages/middleware-jwt/src/spec/jwt.middleware.spec.ts
@@ -16,6 +16,7 @@ describe('JWT middleware', () => {
let utilModule;
let factoryModule;
const injector = createStaticInjectionContainer();
+ const effectMeta = { inject: injector.get };
beforeEach(() => {
jest.unmock('../jwt.util.ts');
@@ -43,7 +44,7 @@ describe('JWT middleware', () => {
utilModule.parseAuthorizationHeader = jest.fn(() => mockedToken);
factoryModule.verifyToken$ = jest.fn(() => () => of(mockedTokenPayload));
- const middleware$ = authorize$({ secret: mockedSecret }, verifyPayload$)(req$, res, injector.get);
+ const middleware$ = authorize$({ secret: mockedSecret }, verifyPayload$)(req$, res, effectMeta);
// then
middleware$.subscribe(
@@ -74,7 +75,7 @@ describe('JWT middleware', () => {
utilModule.parseAuthorizationHeader = jest.fn(() => mockedToken);
factoryModule.verifyToken$ = jest.fn(() => () => throwError(expectedError));
- const middleware$ = authorize$({ secret: mockedSecret }, verifyPayload$)(req$, res, injector.get);
+ const middleware$ = authorize$({ secret: mockedSecret }, verifyPayload$)(req$, res, effectMeta);
// then
middleware$.subscribe(
@@ -106,7 +107,7 @@ describe('JWT middleware', () => {
utilModule.parseAuthorizationHeader = jest.fn(() => mockedToken);
factoryModule.verifyToken$ = jest.fn(() => () => of(mockedTokenPayload));
- const middleware$ = authorize$({ secret: mockedSecret }, verifyPayload$)(req$, res, injector.get);
+ const middleware$ = authorize$({ secret: mockedSecret }, verifyPayload$)(req$, res, effectMeta);
// then
middleware$.subscribe(

MODIFY
packages/websockets/src/operators/mapToServer/mapToServer.operator.spec.ts
packages/websockets/src/operators/mapToServer/mapToServer.operator.spec.ts
import { Marbles, createHttpRequest } from '@marblejs/core/dist/+internal';
-import { ServerEffect, createInjectionToken, ServerEvent, ServerEventType, matchEvent } from '@marblejs/core';
+import { ServerEffect, ServerEvent, ServerEventType, matchEvent } from '@marblejs/core';
import { mapToServer, UpgradeEvent } from './mapToServer.operator';
describe('#mapToServer operator', () => {
let webSocketServerMock;
- let injectorMock;
let socketMock;
let bufferMock;
@@ -15,7 +14,6 @@ describe('#mapToServer operator', () => {
handleUpgrade: jest.fn((req, socket, head, done) => done()),
emit: jest.fn(),
};
- injectorMock = jest.fn(() => webSocketServerMock);
});
test('connects socket to httpServer', () => {
@@ -30,20 +28,20 @@ describe('#mapToServer operator', () => {
};
// when
- const effect$: ServerEffect = (event$, _, inject) =>
+ const effect$: ServerEffect = event$ =>
event$.pipe(
matchEvent(ServerEvent.upgrade),
mapToServer(
- { path: '/test_1', server: createInjectionToken() },
- { path: '/test_2', server: createInjectionToken() },
- )(inject),
+ { path: '/test_1', server: webSocketServerMock },
+ { path: '/test_2', server: webSocketServerMock },
+ ),
);
// then
Marbles.assertEffect(effect$, [
['--a--', { a: incomingEvent }],
['-----', {}],
- ], { meta: injectorMock });
+ ]);
expect(webSocketServerMock.handleUpgrade).toHaveBeenCalledTimes(1);
expect(webSocketServerMock.emit).toHaveBeenCalledTimes(1);
@@ -62,20 +60,20 @@ describe('#mapToServer operator', () => {
};
// when
- const effect$: ServerEffect = (event$, _, inject) =>
+ const effect$: ServerEffect = event$ =>
event$.pipe(
matchEvent(ServerEvent.upgrade),
mapToServer(
- { path: '/test_1', server: createInjectionToken() },
- { path: '/test_2', server: createInjectionToken() },
- )(inject),
+ { path: '/test_1', server: webSocketServerMock },
+ { path: '/test_2', server: webSocketServerMock },
+ ),
);
// then
Marbles.assertEffect(effect$, [
['--a--', { a: incomingEvent }],
['-----', {}],
- ], { meta: injectorMock });
+ ]);
expect(webSocketServerMock.handleUpgrade).not.toHaveBeenCalled();
expect(webSocketServerMock.emit).not.toHaveBeenCalled();

MODIFY
packages/websockets/src/operators/mapToServer/mapToServer.operator.ts
packages/websockets/src/operators/mapToServer/mapToServer.operator.ts
-import { InjectionToken, InjectionGetter, ServerEvent, ServerEventType } from '@marblejs/core';
+import { ServerEvent, ServerEventType } from '@marblejs/core';
import * as pathToRegexp from 'path-to-regexp';
import { Observable, from, EMPTY } from 'rxjs';
import { filter, mergeMap, map, tap, mapTo, toArray, mergeMapTo } from 'rxjs/operators';
@@ -8,16 +8,16 @@ export type UpgradeEvent = ReturnType<typeof ServerEvent.upgrade>;
type WebSocketServerCollection = Array<{
path: string,
- server: InjectionToken,
+ server: MarbleWebSocketServer,
}>;
const isWebSocketUpgrade = ({ request }: UpgradeEvent['payload']) =>
request.headers.upgrade === 'websocket';
-export const mapToServer = (...servers: WebSocketServerCollection) => (inject: InjectionGetter) => {
- const mappedCollection = servers.map(({ path, server: serverToken }) => ({
+export const mapToServer = (...servers: WebSocketServerCollection) => {
+ const mappedCollection = servers.map(({ path, server }) => ({
pathToMatch: pathToRegexp(path),
- serverToken,
+ server,
}));
return (input$: Observable<UpgradeEvent>) =>
@@ -26,10 +26,7 @@ export const mapToServer = (...servers: WebSocketServerCollection) => (inject: I
filter(isWebSocketUpgrade),
mergeMap(({ request, socket, head }) => from(mappedCollection).pipe(
filter(({ pathToMatch }) => pathToMatch.test(request.url!)),
- map(({ serverToken }) => inject<MarbleWebSocketServer>(serverToken)),
- map(data => data),
- filter(Boolean),
- tap(server => server.handleUpgrade(request, socket, head, (ws) =>
+ tap(({ server }) => server.handleUpgrade(request, socket, head, (ws) =>
server.emit(ServerEventType.CONNECTION, ws, request),
)),
mapTo(true),

imp(core, websockets): refactored mapToServer.operator + injector improvements