refactor(http): eagerly loaded `HttpOutputEffect`
MODIFY
packages/http/src/+internal/testing.util.ts
packages/http/src/+internal/testing.util.ts
@@ -84,19 +84,22 @@ export const createMockEffectContext = () => {
return createEffectContext({ ask: lookup(context), client });
};
-export const createTestRoute = (opts?: { throwError?: boolean; delay?: number; method?: HttpMethod }) => {
+export const createTestRoute = (opts?: { throwError?: boolean; delay?: number; method?: HttpMethod; effectSpy?: jest.Mock }) => {
const method = opts?.method ?? 'GET';
const routeDelay = opts?.delay ?? 0;
const req = createHttpRequest(({ url: `/delay_${routeDelay}`, method }));
const path = factorizeRegExpWithParams(`/delay_${routeDelay}`);
- const effect: HttpEffect = req$ =>
- req$.pipe(
+ const effect: HttpEffect = req$ => {
+ opts?.effectSpy?.();
+
+ return req$.pipe(
delay(routeDelay),
tap(() => { if (opts?.throwError) throw new Error(); }),
mapTo({ body: `delay_${routeDelay}` }),
);
+ };
const item: RoutingItem = {
regExp: path.regExp,

MODIFY
packages/http/src/effects/http.effects.interface.ts
packages/http/src/effects/http.effects.interface.ts
@@ -24,7 +24,7 @@ export interface HttpServerEffect<
export interface HttpOutputEffect<
Req extends HttpRequest = HttpRequest,
Res extends HttpEffectResponse = HttpEffectResponse
-> extends HttpEffect<{ req: Req; res: Res }, HttpEffectResponse> {}
+> extends HttpEffect<{ req: Req; res: Res }, { req: Req; res: Res }> {}
export interface HttpEffect<
I = HttpRequest,

MODIFY
packages/http/src/effects/http.requestMetadata.effect.ts
packages/http/src/effects/http.requestMetadata.effect.ts
@@ -10,11 +10,11 @@ export const requestMetadata$: HttpOutputEffect = (out$, ctx) => {
const httpRequestMetadataStorage = useContext(HttpRequestMetadataStorageToken)(ctx.ask);
return out$.pipe(
- map(({ req, res }) => pipe(
- getHttpRequestMetadataIdHeader(req.headers),
- O.fold(constant(res), requestId => {
- httpRequestMetadataStorage.set(requestId, req.meta);
- return res;
+ map(out => pipe(
+ getHttpRequestMetadataIdHeader(out.req.headers),
+ O.fold(constant(out), requestId => {
+ httpRequestMetadataStorage.set(requestId, out.req.meta);
+ return out;
}),
)),
);

MODIFY
packages/http/src/router/http.router.resolver.v2.ts
packages/http/src/router/http.router.resolver.v2.ts
@@ -23,39 +23,35 @@ import { ROUTE_NOT_FOUND_ERROR } from './http.router.effects';
import { decorateEffect } from './http.router.helpers';
import { combineRouteMiddlewares } from './http.router.combiner';
-export const resolveRouting = (
+export const resolveRouting = ({ routing, ctx, output$, error$ } : {
routing: Routing,
ctx: EffectContext<HttpServer>,
-) => (
output$?: HttpOutputEffect,
error$?: HttpErrorEffect,
-) => {
+}) => {
const environmentConfig = provideConfig();
const requestBus = useContext(HttpRequestBusToken)(ctx.ask);
const logger = useContext(LoggerToken)(ctx.ask);
- const close$ = fromEvent(ctx.client, 'close').pipe(take(1), share());
const outputSubject = new Subject<{ res: HttpEffectResponse; req: HttpRequest}>();
const errorSubject = new Subject<{ error: Error; req: HttpRequest }>();
+ const close$ = pipe(
+ fromEvent(ctx.client, 'close'),
+ take(1),
+ share());
+
/**
- * @TODO investigate HttpOutputEffect lazy loading
+ * Outgoing response stream (the result triggers HTTP response call)
*/
- const outputFlow$ = outputSubject.asObservable().pipe(
- mergeMap(data => {
- const stream = environmentConfig.useHttpRequestMetadata() ? requestMetadata$(of(data), ctx) : of(data.res);
- return stream.pipe(map(res => ({ res, req: data.req })));
- }),
- mergeMap(data => {
- const stream = output$ ? output$(of(data), ctx) : of(data.res);
- return stream.pipe(
- map(res => ([res, data.req] as [HttpEffectResponse, HttpRequest])),
- );
- }),
- takeUntil(close$),
- );
+ const response$ = pipe(
+ outputSubject.asObservable(),
+ out$ => environmentConfig.useHttpRequestMetadata() ? requestMetadata$(out$, ctx) : out$,
+ out$ => output$ ? output$(out$, ctx) : out$,
+ takeUntil(close$));
/**
+ * Outgoing error response stream (the result triggers HTTP response call)
* @TODO investigate HttpErrorEffect lazy loading
*/
const errorFlow$ = errorSubject.asObservable().pipe(
@@ -69,9 +65,9 @@ export const resolveRouting = (
takeUntil(close$),
);
- const subscribeOutput = (stream$: Observable<[HttpEffectResponse, HttpRequest]>) =>
+ const subscribeResponse = (stream$: Observable<{ res: HttpEffectResponse, req: HttpRequest }>) =>
stream$
- .pipe(mergeMap(([res, req]) => req.response.send(res)))
+ .pipe(mergeMap(({ res, req }) => req.response.send(res)))
.subscribe({
error: err => { throw unexpectedErrorWhileSendingOutputFactory(err); },
});
@@ -83,7 +79,7 @@ export const resolveRouting = (
error: err => { throw unexpectedErrorWhileSendingErrorFactory(err); },
});
- subscribeOutput(outputFlow$);
+ subscribeResponse(response$);
subscribeError(errorFlow$);
const bootstrappedRrouting: BootstrappedRoutingItem[] = routing.map(item => ({
@@ -176,5 +172,6 @@ export const resolveRouting = (
resolve,
errorSubject,
outputSubject,
+ response$,
};
};

MODIFY
packages/http/src/router/specs/http.router.resolver.v2.spec.ts
packages/http/src/router/specs/http.router.resolver.v2.spec.ts
-import { of, merge, firstValueFrom } from 'rxjs';
+import { Task } from 'fp-ts/lib/Task';
+import { pipe } from 'fp-ts/lib/function';
+import { of, merge, firstValueFrom, lastValueFrom } from 'rxjs';
import { mapTo, take, toArray, delay, mergeMap, map, first } from 'rxjs/operators';
import { createMockEffectContext, createHttpResponse, createHttpRequest, createTestRoute } from '../../+internal/testing.util';
-import { HttpEffect } from '../../effects/http.effects.interface';
+import { HttpEffect, HttpOutputEffect } from '../../effects/http.effects.interface';
import { Routing } from '../http.router.interface';
import { resolveRouting } from '../http.router.resolver.v2';
import { factorizeRegExpWithParams } from '../http.router.params.factory';
@@ -9,7 +11,7 @@ import { HttpError } from '../../error/http.error.model';
import { HttpStatus } from '../../http.interface';
describe('#resolveRouting', () => {
- test('resolves routes inside collection', async (done) => {
+ test('resolves routes inside collection', async done => {
// given
const ctx = createMockEffectContext();
const response = createHttpResponse();
@@ -48,7 +50,7 @@ describe('#resolveRouting', () => {
];
// when
- const { resolve, outputSubject } = resolveRouting(routing, ctx)();
+ const { resolve, outputSubject } = resolveRouting({ routing, ctx });
// then
outputSubject.pipe(take(4), toArray()).subscribe(
@@ -68,6 +70,64 @@ describe('#resolveRouting', () => {
resolve(req5);
});
+ test('resolves `HttpEffect` only once', async () => {
+ // given
+ const effectSpy = jest.fn();
+ const ctx = createMockEffectContext();
+
+ const testData = [
+ createTestRoute({ effectSpy }),
+ ];
+
+ const routing: Routing = testData.map(route => route.item);
+
+ // when
+ const { resolve, response$ } = resolveRouting({ routing, ctx });
+
+ const run: Task<any> = () => {
+ resolve(testData[0].req); // first call
+ resolve(testData[0].req); // second call
+
+ return pipe(response$, take(2), toArray(), lastValueFrom);
+ };
+
+ await run();
+
+ // then
+ expect(effectSpy).toHaveBeenCalledTimes(1);
+ });
+
+ test('resolves `HttpOutputEffect` only once', async () => {
+ // given
+ const effectSpy = jest.fn();
+ const ctx = createMockEffectContext();
+
+ const testData = [
+ createTestRoute(),
+ ];
+
+ const routing: Routing = testData.map(route => route.item);
+
+ const output$: HttpOutputEffect = out$ => (effectSpy(), out$);
+
+ // when
+ const { resolve, response$ } = resolveRouting({ routing, ctx, output$ });
+
+ const run: Task<any> = () => {
+ resolve(testData[0].req); // first call
+ resolve(testData[0].req); // second call
+
+ return pipe(response$, take(2), toArray(), lastValueFrom);
+ };
+
+ await run();
+
+ // then
+ expect(effectSpy).toHaveBeenCalledTimes(1);
+ });
+
+ test.todo('resolves `HttpErrorEffect` only once');
+
test(`returns ${HttpStatus.NOT_FOUND} (Not Found) response if route cannot be resolved`, async () => {
// given
const ctx = createMockEffectContext();
@@ -84,7 +144,7 @@ describe('#resolveRouting', () => {
}];
// when
- const { resolve, errorSubject } = resolveRouting(routing, ctx)();
+ const { resolve, errorSubject } = resolveRouting({ routing, ctx });
const errorPromise = firstValueFrom(errorSubject.pipe(first()));
resolve(req);
@@ -112,7 +172,7 @@ describe('#resolveRouting', () => {
}];
// when
- const { resolve, errorSubject } = resolveRouting(routing, ctx)();
+ const { resolve, errorSubject } = resolveRouting({ routing, ctx });
const errorPromise = firstValueFrom(errorSubject.pipe(first()));
resolve(req);
@@ -147,7 +207,7 @@ describe('#resolveRouting', () => {
}];
// when
- const { resolve, outputSubject } = resolveRouting(routing, ctx)();
+ const { resolve, outputSubject } = resolveRouting({ routing, ctx });
const run = () => {
resolve(testData[0]); // 10 delay
resolve(testData[3]); // 40 delay
@@ -183,7 +243,7 @@ describe('#resolveRouting', () => {
const routing: Routing = testData.map(route => route.item);
// when
- const { resolve, outputSubject } = resolveRouting(routing, ctx)();
+ const { resolve, outputSubject } = resolveRouting({ routing, ctx });
const run = () => {
resolve(testData[0].req); // 10 delay
resolve(testData[3].req); // 40 delay
@@ -219,7 +279,7 @@ describe('#resolveRouting', () => {
const routing: Routing = testData.map(route => route.item);
// when
- const { resolve, outputSubject, errorSubject } = resolveRouting(routing, ctx)();
+ const { resolve, outputSubject, errorSubject } = resolveRouting({ routing, ctx });
const run = () => {
resolve(testData[0].req); // 10 delay
resolve(testData[1].req); // 20 delay

MODIFY
packages/http/src/server/http.server.listener.ts
packages/http/src/server/http.server.listener.ts
@@ -29,10 +29,10 @@ export const httpListener = createListener<HttpListenerConfig, HttpListener>(con
} = config ?? {};
const client = useContext(HttpServerClientToken)(ask);
- const effectContext = createEffectContext({ ask, client });
+ const ctx = createEffectContext({ ask, client });
const routing = factorizeRoutingWithDefaults(effects, middlewares ?? []);
const sendResponse = handleResponse(ask);
- const { resolve } = resolveRouting(routing, effectContext)(output$, error$);
+ const { resolve } = resolveRouting({ routing, ctx, output$, error$ });
const handle = (req: IncomingMessage, res: OutgoingMessage) => {
const marbleReq = req as HttpRequest;

refactor(http): eagerly loaded `HttpOutputEffect`