fix(core): fixed an issue with resubscribed route stream caused by error
MODIFY
packages/@integration/src/http/effects/user.effects.ts
packages/@integration/src/http/effects/user.effects.ts
@@ -3,6 +3,7 @@ import { requestValidator$, t } from '@marblejs/middleware-io';
import { throwError, from, of } from 'rxjs';
import { map, switchMap, catchError, mergeMap, bufferCount } from 'rxjs/operators';
import { Dao } from '../fakes/dao.fake';
+import { simulateRandomFailure, simulateRandomDelay } from '../fakes/random';
import { authorize$ } from '../middlewares/auth.middleware';
const getUserValidator$ = requestValidator$({
@@ -34,11 +35,15 @@ const getUser$ = r.pipe(
return req$.pipe(
use(getUserValidator$),
map(req => req.params.id),
- switchMap(Dao.getUserById),
- map(user => ({ body: user })),
- catchError(() => throwError(
+ mergeMap(id => Dao
+ .getUserById(id)
+ .pipe(catchError(() => throwError(
new HttpError('User does not exist', HttpStatus.NOT_FOUND)
)),
+ )),
+ simulateRandomDelay,
+ simulateRandomFailure,
+ map(user => ({ body: user })),
);
}));

ADD
packages/@integration/src/http/fakes/random.ts
+import { delay, tap } from 'rxjs/operators';
+import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
+
+const getRandomFrom1To10 = () => Math.floor(Math.random() * 10);
+
+export const simulateRandomDelay =
+ delay(isTestEnv() ? 0 : getRandomFrom1To10() * 1000);
+
+export const simulateRandomFailure = tap(() => {
+ const test = getRandomFrom1To10();
+ if (isTestEnv()) return;
+ if (test > 6) throw new Error('Some random error');
+});

MODIFY
packages/core/src/http/router/http.router.combiner.ts
packages/core/src/http/router/http.router.combiner.ts
import { HttpMiddlewareEffect } from '../effects/http.effects.interface';
-import { RouteCombinerConfig, RouteEffectGroup, RouteEffect } from './http.router.interface';
+import { RouteCombinerConfig, RouteEffectGroup, RouteEffect, ErrorSubject } from './http.router.interface';
import { isRouteCombinerConfig, decorateMiddleware } from './http.router.helpers';
export function combineRoutes(path: string, config: RouteCombinerConfig): RouteEffectGroup;
@@ -20,8 +20,8 @@ export function combineRoutes(
}
export const combineRouteMiddlewares =
- (decorate: boolean) => (...middlewares: HttpMiddlewareEffect[]): HttpMiddlewareEffect => (input$, ctx) =>
+ (decorate: boolean, errorSubject: ErrorSubject) => (...middlewares: HttpMiddlewareEffect[]): HttpMiddlewareEffect => (input$, ctx) =>
middlewares.reduce(
- (i$, middleware) => middleware(decorate ? decorateMiddleware(i$) : i$, ctx),
- decorate ? decorateMiddleware(input$) : input$,
+ (i$, middleware) => middleware(decorate ? decorateMiddleware(i$, errorSubject) : i$, ctx),
+ decorate ? decorateMiddleware(input$, errorSubject) : input$,
);

MODIFY
packages/core/src/http/router/http.router.helpers.ts
packages/core/src/http/router/http.router.helpers.ts
-import { Observable, pipe, throwError, of } from 'rxjs';
+import { Observable, pipe, of, EMPTY } from 'rxjs';
import { pipeFromArray } from 'rxjs/internal/util/pipe';
import { mergeMap, catchError, map } from 'rxjs/operators';
import { HttpRequest } from '../http.interface';
import { HttpEffectResponse } from '../effects/http.effects.interface';
-import { isHttpError, HttpError, HttpRequestError } from '../error/http.error.model';
-import { RouteCombinerConfig, RouteEffectGroup } from './http.router.interface';
+import { RouteCombinerConfig, RouteEffectGroup, ErrorSubject } from './http.router.interface';
export const isRouteEffectGroup = (item: any): item is RouteEffectGroup =>
Array.isArray(item.effects) &&
@@ -13,17 +12,16 @@ export const isRouteEffectGroup = (item: any): item is RouteEffectGroup =>
export const isRouteCombinerConfig = (item: any): item is RouteCombinerConfig =>
Array.isArray(item.effects);
-export const decorateEffect = (stream: Observable<HttpRequest>) => {
+export const decorateEffect = (stream: Observable<HttpRequest>, errorSubject: ErrorSubject) => {
stream.pipe = (...operations: any[]): Observable<any> => {
return pipe(
mergeMap((request: HttpRequest) => pipeFromArray([
...operations,
map((res: HttpEffectResponse) => ({ ...res, request })),
- catchError((error: Error) =>
- isHttpError(error)
- ? throwError(new HttpError(error.message, error.status, error.data, request, error.context))
- : throwError(new HttpRequestError(request, error))
- ),
+ catchError((error: Error) => {
+ errorSubject.next({ error, req: request });
+ return EMPTY;
+ }),
])(of(request)) as Observable<HttpEffectResponse>),
)(stream);
};
@@ -31,16 +29,15 @@ export const decorateEffect = (stream: Observable<HttpRequest>) => {
return stream;
};
-export const decorateMiddleware = (stream: Observable<HttpRequest>) => {
+export const decorateMiddleware = (stream: Observable<HttpRequest>, errorSubject: ErrorSubject) => {
stream.pipe = (...operations: any[]): Observable<any> => {
return pipe(
mergeMap((request: HttpRequest) => pipeFromArray([
...operations,
- catchError((error: Error) =>
- isHttpError(error)
- ? throwError(new HttpError(error.message, error.status, error.data, request, error.context))
- : throwError(new HttpRequestError(request, error))
- ),
+ catchError((error: Error) => {
+ errorSubject.next({ error, req: request });
+ return EMPTY;
+ }),
])(of(request)) as Observable<HttpRequest>),
)(stream);
};

MODIFY
packages/core/src/http/router/http.router.interface.ts
packages/core/src/http/router/http.router.interface.ts
@@ -2,6 +2,11 @@ import { Subject } from 'rxjs';
import { HttpMethod, HttpRequest } from '../http.interface';
import { HttpEffect, HttpMiddlewareEffect, HttpEffectResponse } from '../effects/http.effects.interface';
+export type ErrorSubject = Subject<{
+ error: Error;
+ req: HttpRequest<unknown, unknown, unknown>;
+}>;
+
// Route
export interface RouteMeta extends Record<string, any> {
name?: string;

MODIFY
packages/core/src/http/router/http.router.resolver.v2.ts
packages/core/src/http/router/http.router.resolver.v2.ts
import { Subject, of, fromEvent, Observable } from 'rxjs';
-import { takeUntil, share, take, mergeMap, map } from 'rxjs/operators';
+import { takeUntil, share, take, mergeMap, map, catchError } from 'rxjs/operators';
+import { pipe } from 'fp-ts/lib/pipeable';
import { EffectContext } from '../../effects/effects.interface';
import { HttpServer, HttpRequest } from '../http.interface';
import { defaultError$ } from '../error/http.error.effect';
@@ -13,7 +14,7 @@ import {
} from '../error/http.error.model';
import { useContext } from '../../context/context.hook';
import { HttpRequestBusToken } from '../server/http.server.tokens';
-import { LoggerToken, LoggerTag } from '../../logger';
+import { LoggerToken, LoggerTag, LoggerLevel } from '../../logger';
import { Routing, BootstrappedRoutingItem } from './http.router.interface';
import { queryParamsFactory } from './http.router.query.factory';
import { matchRoute } from './http.router.matcher';
@@ -79,6 +80,7 @@ export const resolveRouting = (
const { middlewares, effect, parameters, meta } = methodItem;
const subject = new Subject<HttpRequest>();
const decorate = !meta?.continuous;
+ const middleware = combineRouteMiddlewares(decorate, errorSubject)(...middlewares);
logger({
tag: LoggerTag.HTTP,
@@ -86,10 +88,22 @@ export const resolveRouting = (
message: `Effect mapped: ${item.path || '/'} ${method}`,
})();
- const input$ = subject.asObservable();
- const middleware$ = combineRouteMiddlewares(decorate)(...middlewares)(input$, ctx);
- const effect$ = decorate ? decorateEffect(middleware$) : middleware$;
- const output$ = effect(effect$, ctx).pipe(takeUntil(close$));
+ const output$ = pipe(
+ subject.asObservable(),
+ e$ => middleware(e$, ctx),
+ e$ => decorate ? decorateEffect(e$, errorSubject) : e$,
+ e$ => effect(e$, ctx),
+ e$ => e$.pipe(
+ takeUntil(close$),
+ catchError((error, stream) => processError(stream)(error)),
+ ),
+ );
+
+ const processError = (originStream$: Observable<any>) => (error: any) => {
+ if (!error.request) throw errorNotBoundToRequestErrorFactory(error);
+ errorSubject.next({ error, req: error.request });
+ return originStream$;
+ };
const subscribe = (stream$: Observable<HttpEffectResponse>) =>
stream$.subscribe(
@@ -98,10 +112,15 @@ export const resolveRouting = (
outputSubject.next({ res, req: res.request });
},
error => {
- if (!error.request) throw errorNotBoundToRequestErrorFactory(error);
- errorSubject.next({ error, req: error.request });
- subscribe(stream$);
+ const type = 'RouterResolver';
+ const message = `Unexpected error for Output stream: "${error.name}", "${error.message}"`;
+ logger({ tag: LoggerTag.HTTP, type, message, level: LoggerLevel.ERROR })();
},
+ () => {
+ const type = 'RouterResolver';
+ const message = `Output stream completes`;
+ logger({ tag: LoggerTag.HTTP, type, message, level: LoggerLevel.DEBUG })();
+ }
);
subscribe(output$);

fix(core): fixed an issue with resubscribed route stream caused by error