imp(corer): route resolving optimization v2
MODIFY
packages/core/src/effects/effects.combiner.ts
packages/core/src/effects/effects.combiner.ts
@@ -2,9 +2,9 @@ import { Observable, merge } from 'rxjs';
import { Effect, EffectContext } from './effects.interface';
export const combineMiddlewares = <T, U>
- (...effects: Effect<T, T, U>[]) =>
+ (...effects: (Effect<T, T, U> | undefined)[]) =>
(input$: Observable<T>, ctx: EffectContext<U>): Observable<T> =>
- effects.reduce((i$, effect) => effect(i$, ctx), input$);
+ effects.reduce((i$, effect) => effect ? effect(i$, ctx) : i$, input$);
export const combineEffects = <T, U, V>
(...effects: Effect<T, U, V>[]) =>

MODIFY
packages/core/src/listener/http.listener.ts
packages/core/src/listener/http.listener.ts
import { IncomingMessage, OutgoingMessage } from 'http';
+import { map } from 'rxjs/operators';
import { pipe } from 'fp-ts/lib/pipeable';
import * as R from 'fp-ts/lib/Reader';
import { combineMiddlewares } from '../effects/effects.combiner';
@@ -12,7 +13,7 @@ import { handleResponse } from '../response/response.handler';
import { RouteEffect, RouteEffectGroup, RoutingItem } from '../router/router.interface';
import { resolveRouting } from '../router/router.resolver.v2';
import { factorizeRouting } from '../router/router.factory';
-// import { defaultError$ } from '../error/error.effect';
+import { defaultError$ } from '../error/error.effect';
import { Context, lookup } from '../context/context.factory';
import { createEffectContext } from '../effects/effectsContext.factory';
import { useContext } from '../context/context.hook';
@@ -33,8 +34,8 @@ export interface HttpListener {
export const httpListener = ({
middlewares = [],
effects,
- // error$ = defaultError$,
- // output$ = out$ => out$.pipe(map(o => o.res)),
+ error$ = defaultError$,
+ output$ = out$ => out$.pipe(map(o => o.res)),
}: HttpListenerConfig): R.Reader<Context, HttpListener> => pipe(
R.ask<Context>(),
R.map(ctx => {
@@ -42,9 +43,13 @@ export const httpListener = ({
const ask = lookup(ctx);
const client = useContext(ServerClientToken)(ask);
const effectContext = createEffectContext({ ask, client });
- const combinedMiddlewares = combineMiddlewares(...middlewares);
+ const middleware$ = combineMiddlewares(...middlewares);
const routing = factorizeRouting(effects);
- const provideRoute = resolveRouting(routing, effectContext)(combinedMiddlewares);
+ const provideRoute = resolveRouting(routing, effectContext)(
+ middleware$,
+ output$,
+ error$,
+ );
// const defaultResponse = { status: HttpStatus.NOT_FOUND } as HttpEffectResponse;
const httpServer = (req: IncomingMessage, res: OutgoingMessage) => {
@@ -54,8 +59,13 @@ export const httpListener = ({
marbleRes.send = handleResponse(marbleRes)(marbleReq);
marbleReq.response = marbleRes;
- const subject = provideRoute(marbleReq);
- subject.next(marbleReq);
+ const route = provideRoute(marbleReq);
+
+ if (route.inputSubject) {
+ route.inputSubject.next(marbleReq);
+ } else {
+ route.outputSubject.next({ req: marbleReq, res: { status: 500 }});
+ }
};
httpServer.config = { routing };

MODIFY
packages/core/src/router/router.resolver.v2.ts
packages/core/src/router/router.resolver.v2.ts
-import { EMPTY, Subject, zip } from 'rxjs';
-import { publish} from 'rxjs/operators';
+import { Subject, zip, OperatorFunction, of } from 'rxjs';
+import { publish, catchError, map, mergeMap} from 'rxjs/operators';
import { Routing } from './router.interface';
import { EffectContext } from '../effects/effects.interface';
import { HttpServer, HttpRequest, HttpMethod } from '../http.interface';
import { queryParamsFactory } from './router.query.factory';
-import { HttpMiddlewareEffect } from '../effects/http-effects.interface';
+import { HttpMiddlewareEffect, HttpEffectResponse, HttpErrorEffect, HttpOutputEffect } from '../effects/http-effects.interface';
+import { combineMiddlewares } from '../effects/effects.combiner';
+import { flow } from 'fp-ts/lib/function';
interface RoutingResolver {
- (routing: Routing, ctx: EffectContext<HttpServer>): (middleware$: HttpMiddlewareEffect) =>
- (req: HttpRequest) => Subject<HttpRequest<unknown, unknown, unknown>>;
+ (routing: Routing, ctx: EffectContext<HttpServer>):
+ (middleware$: HttpMiddlewareEffect, output$: HttpOutputEffect, error$: HttpErrorEffect) =>
+ (req: HttpRequest) => {
+ inputSubject?: Subject<HttpRequest<unknown, unknown, unknown>>;
+ outputSubject: Subject<{ res: HttpEffectResponse; req: HttpRequest}>;
+ };
}
interface BootstrappedRoutingItem {
regExp: RegExp;
path: string;
- methods: Partial<Record<HttpMethod, Subject<HttpRequest>>>;
+ methods: Partial<Record<HttpMethod, {
+ input: Subject<HttpRequest>;
+ output: Subject<{ req: HttpRequest; res: HttpEffectResponse }>;
+ }>>;
parameters?: string[] | undefined;
}
type BootstrappedRouting = BootstrappedRoutingItem[];
export interface RouteMatched {
- subject: Subject<HttpRequest>;
+ inputSubject: Subject<HttpRequest>;
+ outputSubject: Subject<{ req: HttpRequest; res: HttpEffectResponse }>;
params: Record<string, string>;
path: string;
}
@@ -45,32 +55,57 @@ export const findRoute = (routing: BootstrappedRouting) => (url: string, method:
}
}
- return { subject, params, path };
+ return { inputSubject: subject.input, outputSubject: subject.output, params, path };
}
return undefined;
};
-export const resolveRouting: RoutingResolver = (routing, ctx) => (globalMiddleware$) => {
+export const resolveRouting: RoutingResolver = (routing, ctx) => (globalMiddleware$, output$, error$) => {
+ const outputSubject = new Subject<{ res: HttpEffectResponse; req: HttpRequest}>();
+ const outputStream$ = outputSubject.asObservable();
+
+ zip(
+ output$(outputStream$, ctx).pipe(catchError(error => of(error))),
+ outputStream$.pipe(map(out => out.req)),
+ ).subscribe(([res, req]) => req.response.send(res));
+
const bootstrappedRrouting: BootstrappedRoutingItem[] = routing.map(item => ({
...item,
methods: Object.entries(item.methods).reduce((acc, [method, methodItem]) => {
if (!methodItem) return { [method]: undefined };
- const req$ = new Subject<HttpRequest>();
-
- const subject$ = req$.pipe(
- publish(req$ => globalMiddleware$(req$, ctx)),
- publish(req$ => methodItem.middleware ? methodItem.middleware(req$, ctx) : req$),
- publish(req$ => zip(
- methodItem.effect(req$, ctx),
- req$,
- )),
+ const isError = (data: any): data is Error =>
+ !!(data as any).stack;
+
+ // eslint-disable-next-line @typescript-eslint/no-unused-vars
+ const errorGuard: OperatorFunction<[HttpEffectResponse, HttpRequest], [HttpEffectResponse, HttpRequest]> = flow(
+ mergeMap(response =>
+ isError(response[0])
+ ? error$(of({ error: response[0], req: response[1] }), ctx)
+ .pipe(map(res => [res, response[1]] as [HttpEffectResponse, HttpRequest]))
+ : of(response)
+ ),
);
- subject$.subscribe(([res, req]) => req.response.send(res));
+ const { middleware, effect } = methodItem;
+
+ const inputSubject = new Subject<HttpRequest>();
+ const inputStream$ = inputSubject.asObservable();
+
+ const middleware$ = combineMiddlewares(globalMiddleware$, middleware);
+
+ zip(
+ inputStream$.pipe(
+ publish(req$ => middleware$(req$, ctx)),
+ publish(req$ => effect(req$, ctx)),
+ catchError(error => of(error)),
+ ),
+ inputStream$,
+ )
+ .subscribe(([res, req]) => outputSubject.next({ req, res }));
- return { ...acc, [method]: subject$ };
+ return { ...acc, [method]: { input: inputSubject, output: outputSubject } };
}, {}),
}));
@@ -82,13 +117,13 @@ export const resolveRouting: RoutingResolver = (routing, ctx) => (globalMiddlewa
const resolvedRoute = find(urlPath, req.method);
- if (!resolvedRoute) { return EMPTY as any as Subject<HttpRequest>; }
+ if (!resolvedRoute) { return { outputSubject }; }
req.query = queryParamsFactory(urlQuery);
req.params = resolvedRoute.params;
req.meta = {};
req.meta.path = resolvedRoute.path;
- return resolvedRoute.subject;
+ return { inputSubject: resolvedRoute.inputSubject, outputSubject: resolvedRoute.outputSubject };
};
};

imp(corer): route resolving optimization v2