feat(core, websockets): Effects output stream
MODIFY
packages/core/src/effects/effects.interface.ts
packages/core/src/effects/effects.interface.ts
@@ -29,6 +29,9 @@ export interface ErrorEffect<T extends Error = HttpError>
export interface ServerEffect<T extends Event = Event>
extends Effect<T, any, http.Server> {}
+export interface OutputEffect<T extends EffectHttpResponse = EffectHttpResponse>
+ extends Effect<T, EffectHttpResponse> {}
+
export interface Effect<
T = HttpRequest,
U = EffectHttpResponse,

MODIFY
packages/core/src/http.interface.ts
packages/core/src/http.interface.ts
@@ -44,11 +44,6 @@ export enum HttpMethodType {
export type HttpMethod = keyof typeof HttpMethodType;
-export type Http = {
- req: HttpRequest;
- res: HttpResponse;
-};
-
export enum HttpStatus {
CONTINUE = 100,
SWITCHING_PROTOCOLS = 101,

MODIFY
packages/core/src/listener/http.listener.ts
packages/core/src/listener/http.listener.ts
import { IncomingMessage, OutgoingMessage } from 'http';
import { of, Subject } from 'rxjs';
-import { catchError, defaultIfEmpty, mergeMap, switchMap, tap, takeWhile } from 'rxjs/operators';
+import { catchError, defaultIfEmpty, mergeMap, tap, takeWhile } from 'rxjs/operators';
import { combineMiddlewares } from '../effects/effects.combiner';
-import { EffectHttpResponse, Middleware, ErrorEffect } from '../effects/effects.interface';
-import { Http, HttpRequest, HttpResponse, HttpStatus } from '../http.interface';
+import { EffectHttpResponse, Middleware, ErrorEffect, OutputEffect } from '../effects/effects.interface';
+import { HttpRequest, HttpResponse, HttpStatus } from '../http.interface';
import { handleResponse } from '../response/response.handler';
import { RouteEffect, RouteEffectGroup } from '../router/router.interface';
import { resolveRouting } from '../router/router.resolver';
@@ -16,39 +16,38 @@ export interface HttpListenerConfig {
middlewares?: Middleware[];
effects: (RouteEffect | RouteEffectGroup)[];
error$?: ErrorEffect;
+ output$?: OutputEffect;
}
export const httpListener = ({
middlewares = [],
effects,
error$ = defaultError$,
+ output$ = out$ => out$,
}: HttpListenerConfig) => {
- const requestSubject$ = new Subject<Http>();
+ const requestSubject$ = new Subject<{ req: HttpRequest; res: HttpResponse; }>();
const combinedMiddlewares = combineMiddlewares(...middlewares);
const routing = factorizeRouting(effects);
const injector = createStaticInjectionContainer();
const defaultMetadata = createEffectMetadata({ inject: injector.get });
const defaultResponse = { status: HttpStatus.NOT_FOUND } as EffectHttpResponse;
- const effect$ = requestSubject$.pipe(
- mergeMap(({ req, res }) => {
- res.send = handleResponse(res)(req);
-
- return combinedMiddlewares(of(req), res, defaultMetadata).pipe(
+ requestSubject$.pipe(
+ tap(({ req, res }) => res.send = handleResponse(res)(req)),
+ mergeMap(({ req, res }) => combinedMiddlewares(of(req), res, defaultMetadata).pipe(
takeWhile(() => !res.finished),
- switchMap(resolveRouting(routing, defaultMetadata)(res)),
+ mergeMap(resolveRouting(routing, defaultMetadata)(res)),
defaultIfEmpty(defaultResponse),
+ mergeMap(out => output$(of(out), res, defaultMetadata)),
tap(res.send),
catchError(error =>
error$(of(req), res, createEffectMetadata({ ...defaultMetadata, error })).pipe(
+ mergeMap(out => output$(of(out), res, createEffectMetadata({ ...defaultMetadata, error }))),
tap(res.send),
),
),
- );
- }),
- );
-
- effect$.subscribe();
+ )),
+ ).subscribe();
const httpServer = (req: IncomingMessage, res: OutgoingMessage) => requestSubject$.next({
req: req as HttpRequest,

MODIFY
packages/core/src/response/responseContentType.factory.ts
packages/core/src/response/responseContentType.factory.ts
@@ -7,12 +7,9 @@ export const DEFAULT_CONTENT_TYPE = ContentType.APPLICATION_JSON;
export const getMimeType = (body: any, path: string) => {
const mimeFromBuffer = Buffer.isBuffer(body) && fileType(body);
-
- if (mimeFromBuffer) {
- return mimeFromBuffer.mime;
- }
-
- return mime.getType(path) || DEFAULT_CONTENT_TYPE;
+ return mimeFromBuffer
+ ? mimeFromBuffer.mime
+ : mime.getType(path) || DEFAULT_CONTENT_TYPE;
};
export const contentTypeFactory = (data: {

MODIFY
packages/websockets/src/effects/ws-effects.interface.ts
packages/websockets/src/effects/ws-effects.interface.ts
@@ -18,6 +18,10 @@ export interface WebSocketConnectionEffect<
T extends http.IncomingMessage = http.IncomingMessage
> extends WebSocketEffect<T, T, MarbleWebSocketClient> {}
+export interface WebSocketOutputEffect<
+ T extends Event = Event
+> extends WebSocketEffect<T, Event> {}
+
export interface WebSocketEffect<
T = Event,
U = Event,

MODIFY
packages/websockets/src/listener/websocket.listener.ts
packages/websockets/src/listener/websocket.listener.ts
@@ -34,6 +34,7 @@ export interface WebSocketListenerConfig {
error$?: WSEffect.WebSocketErrorEffect<Error, any, any>;
eventTransformer?: EventTransformer<Event, any>;
connection$?: WSEffect.WebSocketConnectionEffect;
+ output$?: WSEffect.WebSocketOutputEffect;
}
export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
@@ -43,6 +44,7 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
middlewares = [],
eventTransformer,
connection$ = (req$: Observable<http.IncomingMessage>) => req$,
+ output$ = (out$: Observable<Event>) => out$,
} = config;
const combinedMiddlewares = combineMiddlewares(...middlewares);
@@ -81,9 +83,10 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
const decodedEvent$ = incomingEventSubject$.pipe(map(providedTransformer.decode));
const middlewares$ = combinedMiddlewares(decodedEvent$, client, defaultMetadata);
const effects$ = combinedEffects(eventSubject$, client, defaultMetadata);
+ const effectsOutput$ = output$(effects$, client, defaultMetadata);
let middlewaresSub = subscribeMiddlewares(middlewares$);
- let effectsSub = subscribeEffects(effects$);
+ let effectsSub = subscribeEffects(effectsOutput$);
client.on('message', onMessage);
client.once('close', onClose);

feat(core, websockets): Effects output stream