reafactor(core, websockets): injector refactor
MODIFY
packages/@integration/src/index.ts
packages/@integration/src/index.ts
@@ -27,7 +27,7 @@ export const server = createServer({
port: 1337,
httpListener: httpServer,
dependencies: [
- bind(WebSocketsToken).to(() => webSocketServer()),
+ bind(WebSocketsToken).to(webSocketServer({ noServer: true })),
],
event$: (...args) => merge(
listen$(...args),

MODIFY
packages/core/src/index.ts
packages/core/src/index.ts
@@ -12,5 +12,6 @@ export * from './event/event.factory';
export * from './event/event.interface';
export * from './server/server.injector';
export * from './server/server.event';
+export * from './server/server.token';
export { createServer, CreateServerConfig } from './server/server.factory';
export { HttpListenerConfig, httpListener } from './listener/http.listener';

MODIFY
packages/core/src/server/server.factory.ts
packages/core/src/server/server.factory.ts
@@ -5,6 +5,7 @@ import { isCloseEvent } from './server.event';
import { subscribeServerEvents } from './server.event.subscriber';
import { InjectionDependencies } from './server.injector';
import { ServerEffect } from '../effects/effects.interface';
+import { httpServerToken } from './server.token';
export interface CreateServerConfig {
port?: number;
@@ -22,8 +23,10 @@ export const createServer = (config: CreateServerConfig) => {
const server = http.createServer(httpListener);
const serverEvent$ = eventsSubscriber(server).pipe(takeWhile(e => !isCloseEvent(e)));
+ injector.register(httpServerToken, () => server);
+
if (dependencies) {
- injector.registerAll(dependencies)(server);
+ injector.registerAll(dependencies);
}
if (event$) {

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 Injectable = (context: Injector) => any;
export type InjectionDependencies = { token: InjectionToken, factory: Injectable }[];
export type InjectionToken<T = any> = new() => Token<T>;
@@ -15,22 +13,31 @@ export class Token<T = any> {
export class StaticInjector {
private dependencies = new Map<InjectionToken, any>();
- register = <T>(token: InjectionToken<T>, factory: Injectable) => (httpServer: http.Server) =>
- this.dependencies.set(token, factory(httpServer))
+ register = <T>(token: InjectionToken<T>, factory: Injectable) => {
+ this.dependencies.set(token, factory(this));
+ return this;
+ }
- deregister = <T>(token: InjectionToken<T>) =>
- this.dependencies.delete(token)
+ deregister = <T>(token: InjectionToken<T>) => {
+ this.dependencies.delete(token);
+ return this;
+ }
- get = <T>(token: InjectionToken<T>): T =>
- this.dependencies.get(token)
+ get = <T>(token: InjectionToken<T>): T => {
+ return this.dependencies.get(token);
+ }
- registerAll = (dependenciesToRegister: InjectionDependencies) => (httpServer: http.Server) =>
+ registerAll = (dependenciesToRegister: InjectionDependencies) => {
dependenciesToRegister.forEach(({ token, factory }) =>
- this.register(token, factory)(httpServer)
- )
+ this.register(token, factory)
+ );
+ return this;
+ }
- deregisterAll = () =>
- this.dependencies.clear()
+ deregisterAll = () => {
+ this.dependencies.clear();
+ return this;
+ }
}
export const createInjectionToken = <T>() =>
@@ -39,6 +46,8 @@ export const createInjectionToken = <T>() =>
export const createStaticInjectionContainer = () =>
new StaticInjector();
+export type Injector = StaticInjector;
+
export const bind = <T>(token: InjectionToken<T>) => ({
to: (factory: Injectable) => ({ token, factory })
});

ADD
packages/core/src/server/server.token.ts
+import * as http from 'http';
+import { createInjectionToken } from './server.injector';
+
+export const httpServerToken = createInjectionToken<http.Server>();

MODIFY
packages/core/src/server/specs/server.injector.spec.ts
packages/core/src/server/specs/server.injector.spec.ts
-import * as http from 'http';
import { bind, createInjectionToken, createStaticInjectionContainer } from '../server.injector';
describe('#bind', () => {
- const httpServer = {} as http.Server;
+ const injector = createStaticInjectionContainer();
test('factorizes dependency', () => {
// given
@@ -14,13 +13,11 @@ describe('#bind', () => {
// then
expect(factorizedDependency.token).toBe(Token);
- expect(factorizedDependency.factory(httpServer)).toEqual(dependency);
+ expect(factorizedDependency.factory(injector)).toEqual(dependency);
});
});
describe('#createStaticInjectionContainer', () => {
- const httpServer = {} as http.Server;
-
test('#register registers single dependency', () => {
// given
const dependency = { test: 'test_string' };
@@ -28,7 +25,7 @@ describe('#createStaticInjectionContainer', () => {
const container = createStaticInjectionContainer();
// when
- container.register(Token, () => dependency)(httpServer);
+ container.register(Token, () => dependency);
// then
expect(container.get(Token)).toEqual(dependency);
@@ -49,7 +46,7 @@ describe('#createStaticInjectionContainer', () => {
{ token: Token1, factory: () => dependency1 },
{ token: Token2, factory: () => dependency2 },
{ token: Token3, factory: () => dependency3 },
- ])(httpServer);
+ ]);
// then
expect(container.get(Token1)).toEqual(dependency1);
@@ -66,8 +63,8 @@ describe('#createStaticInjectionContainer', () => {
const container = createStaticInjectionContainer();
// when
- container.register(Token1, () => dependency1)(httpServer);
- container.register(Token2, () => dependency2)(httpServer);
+ container.register(Token1, () => dependency1);
+ container.register(Token2, () => dependency2);
container.deregister(Token1);
// then
@@ -82,7 +79,7 @@ describe('#createStaticInjectionContainer', () => {
const container = createStaticInjectionContainer();
// when
- container.register(Token, () => dependency)(httpServer);
+ container.register(Token, () => dependency);
container.deregisterAll();
// then

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 { app } from './io-ws.integration';
+import { createStaticInjectionContainer, httpServerToken } from '@marblejs/core';
describe('@marblejs/middleware-io - WebSocket integration', () => {
const testBed = createWebSocketsTestBed();
@@ -13,9 +14,10 @@ describe('@marblejs/middleware-io - WebSocket integration', () => {
const event = JSON.stringify({ type: 'POST_USER', payload: user });
const httpServer = testBed.getServer();
const targetClient = testBed.getClient();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
// when
- app(httpServer);
+ app()(injector);
targetClient.once('open', () => targetClient.send(event));
// then
@@ -29,6 +31,7 @@ describe('@marblejs/middleware-io - WebSocket integration', () => {
// given
const httpServer = testBed.getServer();
const targetClient = testBed.getClient();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
const user = { id: 'id', age: '100', };
const event = JSON.stringify({ type: 'POST_USER', payload: user });
const expectedError = {
@@ -40,7 +43,7 @@ describe('@marblejs/middleware-io - WebSocket integration', () => {
};
// when
- app(httpServer);
+ app()(injector);
targetClient.once('open', () => targetClient.send(event));
// then

MODIFY
packages/websockets/src/listener/specs/websocket.listener.spec.ts
packages/websockets/src/listener/specs/websocket.listener.spec.ts
-import { Event, EventError } from '@marblejs/core';
+import { Event, EventError, createStaticInjectionContainer, httpServerToken } from '@marblejs/core';
import { throwError, fromEvent, forkJoin } from 'rxjs';
import { tap, map, mergeMap, first, toArray, take } from 'rxjs/operators';
import { webSocketListener } from '../websocket.listener';
@@ -18,12 +18,13 @@ describe('WebSocket listener', () => {
// given
const targetClient = testBed.getClient(0);
const httpServer = testBed.getServer();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
const echo$: WebSocketEffect = event$ => event$;
const event = JSON.stringify({ type: 'EVENT', payload: 'test' });
const webSocketServer = webSocketListener({ effects: [echo$] });
// when
- webSocketServer(httpServer);
+ webSocketServer()(injector);
targetClient.once('open', () => targetClient.send(event));
// then
@@ -41,10 +42,11 @@ describe('WebSocket listener', () => {
const event = JSON.stringify({ type: 'EVENT', payload: 'test' });
const webSocketServer = webSocketListener({ effects: [echo$] });
const httpServer = testBed.getServer();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
const targetClient = testBed.getClient(0);
// when
- webSocketServer(httpServer);
+ webSocketServer()(injector);
targetClient.on('open', () => targetClient.send(event));
// then
@@ -62,9 +64,10 @@ describe('WebSocket listener', () => {
// given
const echo$: WebSocketEffect = event$ => event$;
const event = JSON.stringify({ type: 'EVENT', payload: 'test' });
- const webSocketServer = webSocketListener({ effects: [echo$] })();
- const targetClient = testBed.getClient(0);
const httpServer = testBed.getServer();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
+ const webSocketServer = webSocketListener({ effects: [echo$] })({ noServer: true })(injector);
+ const targetClient = testBed.getClient(0);
// when
httpServer.on('upgrade', (request, socket, head) => {
@@ -93,13 +96,14 @@ describe('WebSocket listener', () => {
);
const targetClient = testBed.getClient(0);
const httpServer = testBed.getServer();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
const webSocketServer = webSocketListener({
effects: [e$],
middlewares: [m$, m$, m$],
});
// when
- webSocketServer(httpServer);
+ webSocketServer()(injector);
targetClient.once('open', () => targetClient.send(incomingEvent));
// then
@@ -118,10 +122,11 @@ describe('WebSocket listener', () => {
});
const targetClient = testBed.getClient(0);
const httpServer = testBed.getServer();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
const webSocketServer = webSocketListener();
// when
- webSocketServer(httpServer);
+ webSocketServer()(injector);
targetClient.once('open', () => {
targetClient.send(incomingEvent);
targetClient.send(incomingEvent);
@@ -146,10 +151,11 @@ describe('WebSocket listener', () => {
);
const targetClient = testBed.getClient(0);
const httpServer = testBed.getServer();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
const webSocketServer = webSocketListener({ effects: [effect$] });
// when
- webSocketServer(httpServer);
+ webSocketServer()(injector);
targetClient.once('open', () => {
targetClient.send(incomingEvent);
targetClient.send(incomingEvent);
@@ -172,9 +178,10 @@ describe('WebSocket listener', () => {
const webSocketServer = webSocketListener({ connection$ });
const targetClient = testBed.getClient(0);
const httpServer = testBed.getServer();
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
// when
- webSocketServer(httpServer);
+ webSocketServer()(injector);
targetClient.once('open', () => targetClient.send('test'));
// then
@@ -205,9 +212,10 @@ describe('WebSocket listener', () => {
);
const httpServer = testBed.getServer();
const webSocketServer = webSocketListener({ effects: [effect$], eventTransformer });
+ const injector = createStaticInjectionContainer().register(httpServerToken, () => httpServer);
// when
- webSocketServer(httpServer);
+ webSocketServer()(injector);
targetClient.once('open', () => {
targetClient.send(Buffer.from(decodedMessage));
});

MODIFY
packages/websockets/src/listener/websocket.listener.ts
packages/websockets/src/listener/websocket.listener.ts
@@ -5,9 +5,11 @@ import { map } from 'rxjs/operators';
import {
combineEffects,
combineMiddlewares,
- createStaticInjectionContainer,
Event,
EffectMetadata,
+ httpServerToken,
+ Injector,
+ Injectable,
} from '@marblejs/core';
import * as WS from '../websocket.interface';
import * as WSHelper from './websocket.helper';
@@ -19,11 +21,11 @@ import { handleEffectsError } from '../error/ws-error.handler';
import { provideErrorEffect } from '../error/ws-error.provider';
type HandleIncomingMessage =
- (client: WS.MarbleWebSocketClient) =>
+ (client: WS.MarbleWebSocketClient, context: Injector) =>
() => void;
type HandleIncomingConnection =
- (server: WS.MarbleWebSocketServer) =>
+ (server: WS.MarbleWebSocketServer, context: Injector) =>
(client: WS.WebSocketClient, request: http.IncomingMessage) => void;
export interface WebSocketListenerConfig {
@@ -47,10 +49,8 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
const combinedEffects = combineEffects(...effects);
const providedError$ = provideErrorEffect(error$, eventTransformer);
const providedTransformer: EventTransformer<any, any> = eventTransformer || jsonTransformer;
- const injector = createStaticInjectionContainer();
- const defaultMetadata: EffectMetadata = { inject: injector.get };
- const handleIncomingMessage: HandleIncomingMessage = client => () => {
+ const handleIncomingMessage: HandleIncomingMessage = (client, ctx) => () => {
const subscribeMiddlewares = (input$: Observable<any>) =>
input$.subscribe(
event => eventSubject$.next(event),
@@ -75,6 +75,7 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
effectsSub.unsubscribe();
};
+ const defaultMetadata: EffectMetadata = { inject: ctx.get };
const incomingEventSubject$ = new Subject<WS.WebSocketData>();
const eventSubject$ = new Subject<any>();
const decodedEvent$ = incomingEventSubject$.pipe(map(providedTransformer.decode));
@@ -88,8 +89,9 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
client.once('close', onClose);
};
- const handleIncomingConnection: HandleIncomingConnection = (server) => (client, req) => {
+ const handleIncomingConnection: HandleIncomingConnection = (server, ctx) => (client, req) => {
const request$ = of(req);
+ const defaultMetadata: EffectMetadata = { inject: ctx.get };
const extendedClient = WSHelper.extendClientWith({
sendResponse: handleResponse(client, providedTransformer),
sendBroadcastResponse: handleBroadcastResponse(server, providedTransformer),
@@ -97,20 +99,20 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
})(client);
connection$(request$, extendedClient, defaultMetadata).subscribe(
- handleIncomingMessage(extendedClient),
+ handleIncomingMessage(extendedClient, ctx),
WSHelper.handleClientValidationError(extendedClient),
);
WSHelper.handleClientBrokenConnection(extendedClient).subscribe();
};
- return (server?: http.Server) => {
- const serverOptions: WebSocket.ServerOptions = server ? { server } : { noServer: true };
- const webSocketServer = WSHelper.createWebSocketServer(serverOptions);
+ return (serverOptions?: WebSocket.ServerOptions): Injectable => ctx => {
+ const providedOptions: WebSocket.ServerOptions = serverOptions || { server: ctx.get(httpServerToken) };
+ const webSocketServer = WSHelper.createWebSocketServer(providedOptions);
const sendBroadcastResponse = handleBroadcastResponse(webSocketServer, providedTransformer);
const extendedWebSocketServer = WSHelper.extendServerWith({ sendBroadcastResponse })(webSocketServer);
- extendedWebSocketServer.on('connection', handleIncomingConnection(extendedWebSocketServer));
+ extendedWebSocketServer.on('connection', handleIncomingConnection(extendedWebSocketServer, ctx));
WSHelper.handleServerBrokenConnections(extendedWebSocketServer).subscribe();
return extendedWebSocketServer;

reafactor(core, websockets): injector refactor