feat(core, websockets): Effects output stream

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,

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,

packages/core/src/listener/http.listener.ts

@@ -1,9 +1,9 @@
 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(
-        takeWhile(() => !res.finished),
-        switchMap(resolveRouting(routing, defaultMetadata)(res)),
-        defaultIfEmpty(defaultResponse),
-        tap(res.send),
-        catchError(error =>
-          error$(of(req), res, createEffectMetadata({ ...defaultMetadata, error })).pipe(
-            tap(res.send),
-          ),
+  requestSubject$.pipe(
+    tap(({ req, res }) => res.send = handleResponse(res)(req)),
+    mergeMap(({ req, res }) => combinedMiddlewares(of(req), res, defaultMetadata).pipe(
+      takeWhile(() => !res.finished),
+      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,

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: {

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,

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);