refactor(core, websockets): Effect metadata
MODIFY
packages/@integration/src/effects/calculator.ws-effects.ts
packages/@integration/src/effects/calculator.ws-effects.ts
@@ -8,11 +8,11 @@ export const sum$: WebSocketEffect = event$ =>
matchEvent('SUM')
);
-export const add$: WebSocketEffect = (event$, client) =>
+export const add$: WebSocketEffect = (event$, ...args) =>
event$.pipe(
matchEvent('ADD'),
use(eventValidator$(t.number)),
- buffer(sum$(event$, client)),
+ buffer(sum$(event$, ...args)),
map(events => events.reduce((a, e) => e.payload! + a, 0)),
map(payload => ({ type: 'SUM_RESULT', payload })),
);

MODIFY
packages/core/src/effects/effects.combiner.ts
packages/core/src/effects/effects.combiner.ts
import { Observable, merge } from 'rxjs';
-import { Effect } from './effects.interface';
+import { Effect, EffectMetadata } from './effects.interface';
export const combineMiddlewares = <T, U>
(...effects: Effect<T, T, U>[]) =>
- (input$: Observable<T>, client: U, meta?: any): Observable<T> =>
- effects.reduce((i$, effect) => effect(i$, client, meta), input$);
+ (input$: Observable<T>, client: U, meta: EffectMetadata): Observable<T> =>
+ effects.reduce((i$, effect) => effect(i$, client, meta!), input$);
export const combineEffects = <T, U, V>
(...effects: Effect<T, U, V>[]) =>
- (input$: Observable<T>, client: V, meta?: any): Observable<U> =>
+ (input$: Observable<T>, client: V, meta: EffectMetadata): Observable<U> =>
merge(...effects.map(effect => effect(input$, client, meta)));

MODIFY
packages/core/src/effects/effects.interface.ts
packages/core/src/effects/effects.interface.ts
@@ -5,9 +5,9 @@ import { HttpError } from '../error/error.model';
import { InjectorGetter } from '../server/server.injector';
import { Event } from '../event/event.interface';
-export interface EffectMetadata<T extends Error> {
- error?: T;
+export interface EffectMetadata<T extends Error = Error> {
inject: InjectorGetter;
+ error?: T;
[key: string]: any;
}

MODIFY
packages/core/src/listener/http.listener.ts
packages/core/src/listener/http.listener.ts
@@ -27,19 +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 defaultMetadata = { inject: injector.get };
const effect$ = requestSubject$.pipe(
mergeMap(({ req, res }) => {
res.send = handleResponse(res)(req);
- return combinedMiddlewares(of(req), res, effectDefaultMeta).pipe(
+ return combinedMiddlewares(of(req), res, defaultMetadata).pipe(
takeWhile(() => !res.finished),
- switchMap(resolveRouting(routing, injector.get)(res)),
+ switchMap(resolveRouting(routing, defaultMetadata)(res)),
defaultIfEmpty(defaultResponse),
tap(res.send),
catchError(error =>
- error$(of(req), res, { ...effectDefaultMeta, error }).pipe(
+ error$(of(req), res, { ...defaultMetadata, error }).pipe(
tap(res.send),
),
),

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

MODIFY
packages/core/src/router/specs/router.resolver.spec.ts
packages/core/src/router/specs/router.resolver.spec.ts
import { mapTo, tap, map } from 'rxjs/operators';
-import { Effect, Middleware } from '../../effects/effects.interface';
+import { Effect, Middleware, EffectMetadata } from '../../effects/effects.interface';
import { findRoute, resolveRouting } from '../router.resolver';
import { HttpRequest, HttpResponse, HttpMethod } from '../../http.interface';
import { RouteMatched, Routing } from '../router.interface';
@@ -99,13 +99,14 @@ describe('#findRoute', () => {
describe('#resolveRouting', () => {
let router;
let queryFactory;
- const injector = createStaticInjectionContainer();
+ let metadata: EffectMetadata;
beforeEach(() => {
jest.unmock('../router.resolver.ts');
jest.unmock('../router.query.factory');
router = require('../router.resolver.ts');
queryFactory = require('../router.query.factory');
+ metadata = { inject: createStaticInjectionContainer().get };
});
test('resolves found effect', done => {
@@ -118,7 +119,7 @@ describe('#resolveRouting', () => {
// when
router.findRoute = jest.fn(() => expectedMachingResult);
queryFactory.queryParamsFactory = jest.fn(() => ({}));
- const resolvedRoute = resolveRouting([], injector.get)(res)(req);
+ const resolvedRoute = resolveRouting([], metadata)(res)(req);
// then
resolvedRoute.subscribe(effect => {
@@ -136,7 +137,7 @@ describe('#resolveRouting', () => {
// when
router.findRoute = jest.fn(() => undefined);
- const resolvedRoute = resolveRouting([], injector.get)(res)(req);
+ const resolvedRoute = resolveRouting([], metadata)(res)(req);
// then
resolvedRoute.subscribe(
@@ -161,7 +162,7 @@ describe('#resolveRouting', () => {
const res = { finished: true } as HttpResponse;
// when
- const resolvedRoute = resolveRouting([], injector.get)(res)(req);
+ const resolvedRoute = resolveRouting([], metadata)(res)(req);
// then
resolvedRoute.subscribe(
@@ -192,7 +193,7 @@ describe('#resolveRouting', () => {
// when
router.findRoute = jest.fn(() => expectedMachingResult);
queryFactory.queryParamsFactory = jest.fn(() => ({}));
- const resolvedRoute = resolveRouting([], injector.get)(res)(req);
+ const resolvedRoute = resolveRouting([], metadata)(res)(req);
// then
resolvedRoute.subscribe(effect => {

MODIFY
packages/middleware-joi/test/unit/body.spec.ts
packages/middleware-joi/test/unit/body.spec.ts
@@ -5,7 +5,7 @@ import { bodyParser$ } from '@marblejs/middleware-body';
const MockReq = require('mock-req');
describe('Joi middleware - Body', () => {
- const injector = createStaticInjectionContainer();
+ const metadata = { inject: createStaticInjectionContainer().get };
it('should throws an error if dont pass a required field', done => {
expect.assertions(2);
@@ -24,7 +24,7 @@ describe('Joi middleware - Body', () => {
})
};
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, metadata);
const valid$ = validator$(schema)(http$);
valid$.subscribe(
@@ -60,7 +60,7 @@ describe('Joi middleware - Body', () => {
})
};
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, metadata);
const valid$ = validator$(schema)(http$);
valid$.subscribe(
@@ -96,7 +96,7 @@ describe('Joi middleware - Body', () => {
})
};
- const http$ = bodyParser$(req$, res, injector.get);
+ const http$ = bodyParser$(req$, res, metadata);
const valid$ = validator$(schema, { allowUnknown: true })(http$);
valid$.subscribe(data => {

MODIFY
packages/websockets/src/effects/ws-effects.interface.ts
packages/websockets/src/effects/ws-effects.interface.ts
import * as http from 'http';
-import { Event } from '@marblejs/core';
+import { Event, EffectMetadata } from '@marblejs/core';
import { Observable } from 'rxjs';
import { MarbleWebSocketClient } from '../websocket.interface';
@@ -22,7 +22,7 @@ export interface WebSocketEffect<
T = Event,
U = Event,
V = MarbleWebSocketClient,
- W = any,
+ 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/websockets/src/error/specs/ws-error.effect.spec.ts
packages/websockets/src/error/specs/ws-error.effect.spec.ts
@@ -19,7 +19,7 @@ describe('error$', () => {
Marbles.assertEffect(error$, [
['--a--', { a: incomingEvent }],
['--b--', { b: outgoingEvent }],
- ], { meta: error });
+ ], { meta: { error } });
});
test('returns stream of error events for undefined error object', () => {
@@ -32,7 +32,7 @@ describe('error$', () => {
Marbles.assertEffect(error$, [
['--a--', { a: incomingEvent }],
['--b--', { b: outgoingEvent }],
- ], { meta: error });
+ ], { meta: { error } });
});
test('returns stream of error events for undefined event object', () => {
@@ -45,6 +45,6 @@ describe('error$', () => {
Marbles.assertEffect(error$, [
['--a--', { a: incomingEvent }],
['--b--', { b: outgoingEvent }],
- ], { meta: error });
+ ], { meta: { error } });
});
});

MODIFY
packages/websockets/src/error/specs/ws-error.handler.spec.ts
packages/websockets/src/error/specs/ws-error.handler.spec.ts
-import { EventError } from '@marblejs/core';
+import { EventError, createStaticInjectionContainer, EffectMetadata } from '@marblejs/core';
import { mapTo } from 'rxjs/operators';
import { handleEffectsError } from '../ws-error.handler';
import { MarbleWebSocketClient } from '../../websocket.interface';
import { WebSocketErrorEffect } from '../../effects/ws-effects.interface';
describe('#handleEffectsError', () => {
+ let defaultMetadata: EffectMetadata;
+
+ beforeEach(() => {
+ defaultMetadata = { inject: createStaticInjectionContainer().get };
+ });
+
test('handles error if error$ is defined', () => {
// given
const client = { sendResponse: jest.fn() } as any as MarbleWebSocketClient;
@@ -14,7 +20,7 @@ describe('#handleEffectsError', () => {
);
// when
- handleEffectsError(client, error$)(error);
+ handleEffectsError(defaultMetadata, client, error$)(error);
// then
expect(client.sendResponse).toHaveBeenCalled();
@@ -26,7 +32,7 @@ describe('#handleEffectsError', () => {
const error = new EventError({ type: 'EVENT' }, '');
// when
- handleEffectsError(client, undefined)(error);
+ handleEffectsError(defaultMetadata, client, undefined)(error);
// then
expect(client.sendResponse).not.toHaveBeenCalled();

MODIFY
packages/websockets/src/error/ws-error.effect.ts
packages/websockets/src/error/ws-error.effect.ts
@@ -8,7 +8,7 @@ const errorFactory = (message: string | undefined, data: any | undefined) => ({
message, data,
});
-export const error$: WebSocketErrorEffect<EventError> = (event$, _, error) =>
+export const error$: WebSocketErrorEffect<EventError> = (event$, _, { error }) =>
event$.pipe(
map(event => ({
type: event ? event.type : DEFAULT_ERROR_CHANNEL,

MODIFY
packages/websockets/src/error/ws-error.handler.ts
packages/websockets/src/error/ws-error.handler.ts
import { of } from 'rxjs';
-import { EventError } from '@marblejs/core';
+import { EventError, EffectMetadata } from '@marblejs/core';
import { MarbleWebSocketClient } from '../websocket.interface';
import { WebSocketErrorEffect } from '../effects/ws-effects.interface';
export const handleEffectsError = <IncomingError extends EventError>(
+ metadata: EffectMetadata,
client: MarbleWebSocketClient,
error$: WebSocketErrorEffect<IncomingError, any, any> | undefined,
) => (error: IncomingError) => {
if (error$) {
- error$(of(error.event), client, error).subscribe(client.sendResponse);
+ error$(of(error.event), client, { ...metadata, error }).subscribe(client.sendResponse);
}
};

MODIFY
packages/websockets/src/listener/websocket.listener.ts
packages/websockets/src/listener/websocket.listener.ts
@@ -2,7 +2,13 @@ import * as http from 'http';
import * as WebSocket from 'ws';
import { Subject, of, Observable } from 'rxjs';
import { map } from 'rxjs/operators';
-import { combineEffects, combineMiddlewares, Event } from '@marblejs/core';
+import {
+ combineEffects,
+ combineMiddlewares,
+ createStaticInjectionContainer,
+ Event,
+ EffectMetadata,
+} from '@marblejs/core';
import * as WS from '../websocket.interface';
import * as WSHelper from './websocket.helper';
import * as WSEffect from '../effects/ws-effects.interface';
@@ -41,18 +47,20 @@ 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 subscribeMiddlewares = (input$: Observable<any>) =>
input$.subscribe(
event => eventSubject$.next(event),
- error => handleEffectsError(client, providedError$)(error),
+ error => handleEffectsError(defaultMetadata, client, providedError$)(error),
);
const subscribeEffects = (input$: Observable<any>) =>
input$.subscribe(
event => client.sendResponse(event),
- error => handleEffectsError(client, providedError$)(error),
+ error => handleEffectsError(defaultMetadata, client, providedError$)(error),
);
const onMessage = (event: WS.WebSocketData) => {
@@ -70,8 +78,8 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
const incomingEventSubject$ = new Subject<WS.WebSocketData>();
const eventSubject$ = new Subject<any>();
const decodedEvent$ = incomingEventSubject$.pipe(map(providedTransformer.decode));
- const middlewares$ = combinedMiddlewares(decodedEvent$, client);
- const effects$ = combinedEffects(eventSubject$, client);
+ const middlewares$ = combinedMiddlewares(decodedEvent$, client, defaultMetadata);
+ const effects$ = combinedEffects(eventSubject$, client, defaultMetadata);
let middlewaresSub = subscribeMiddlewares(middlewares$);
let effectsSub = subscribeEffects(effects$);
@@ -88,7 +96,7 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
isAlive: true,
})(client);
- connection$(request$, extendedClient).subscribe(
+ connection$(request$, extendedClient, defaultMetadata).subscribe(
handleIncomingMessage(extendedClient),
WSHelper.handleClientValidationError(extendedClient),
);

refactor(core, websockets): Effect metadata