fix(http): collect request metadata in #subscribeResponse
DELETE
packages/http/src/effects/http.requestMetadata.effect.ts
-import { useContext } from '@marblejs/core';
-import * as O from 'fp-ts/lib/Option';
-import { constant, pipe } from 'fp-ts/lib/function';
-import { map } from 'rxjs/operators';
-import { HttpRequestMetadataStorageToken } from '../server/internal-dependencies/httpRequestMetadataStorage.reader';
-import { getHttpRequestMetadataIdHeader } from '../+internal/metadata.util';
-import { HttpOutputEffect } from './http.effects.interface';
-
-export const requestMetadata$: HttpOutputEffect = (output$, ctx) => {
- const httpRequestMetadataStorage = useContext(HttpRequestMetadataStorageToken)(ctx.ask);
-
- return output$.pipe(
- map(response => pipe(
- getHttpRequestMetadataIdHeader(response.request.headers),
- O.fold(constant(response), requestId => {
- httpRequestMetadataStorage.set(requestId, response.request.meta);
- return response;
- }),
- )),
- );
-};

MODIFY
packages/http/src/router/http.router.resolver.ts
packages/http/src/router/http.router.resolver.ts
import { Observable, Subject, fromEvent, merge } from 'rxjs';
-import { takeUntil, share, take, mergeMap, map, catchError } from 'rxjs/operators';
-import { flow, pipe } from 'fp-ts/lib/function';
+import { takeUntil, share, take, mergeMap, map, catchError, tap } from 'rxjs/operators';
+import { constVoid, flow, pipe } from 'fp-ts/lib/function';
+import * as O from 'fp-ts/lib/Option';
import { EffectContext, useContext, LoggerToken, LoggerTag, LoggerLevel } from '@marblejs/core';
import { throwException, getErrorMessage } from '@marblejs/core/dist/+internal/utils';
import { HttpServer, HttpRequest, HttpStatus, WithHttpRequest } from '../http.interface';
@@ -15,8 +16,7 @@ import {
isURIError,
} from '../error/http.error.model';
import { HttpRequestBusToken } from '../server/internal-dependencies/httpRequestBus.reader';
-import { requestMetadata$ } from '../effects/http.requestMetadata.effect';
-import { provideConfig } from '../http.config';
+import { HttpRequestMetadataStorageToken } from '../server/internal-dependencies/httpRequestMetadataStorage.reader';
import { Routing, BootstrappedRoutingItem } from './http.router.interface';
import { queryParamsFactory } from './http.router.query.factory';
import { matchRoute } from './http.router.matcher';
@@ -32,11 +32,11 @@ type ResolveRoutingConfig = Readonly<{
}>
export const resolveRouting = (config: ResolveRoutingConfig) => {
- const environmentConfig = provideConfig();
const requestBus = useContext(HttpRequestBusToken)(config.ctx.ask);
const logger = useContext(LoggerToken)(config.ctx.ask);
const outputSubject = new Subject<WithHttpRequest<HttpEffectResponse>>();
const errorSubject = new Subject<WithHttpRequest<{ error: Error }>>();
+ const requestMetadataStorage = config.ctx.ask(HttpRequestMetadataStorageToken);
/**
* Server close stream (closes all active streams)
@@ -51,7 +51,6 @@ export const resolveRouting = (config: ResolveRoutingConfig) => {
*/
const response$ = pipe(
outputSubject.asObservable(),
- o$ => environmentConfig.useHttpRequestMetadata() ? requestMetadata$(o$, config.ctx) : o$,
o$ => config.output$ ? config.output$(o$, config.ctx) : o$,
takeUntil(close$),
);
@@ -73,7 +72,11 @@ export const resolveRouting = (config: ResolveRoutingConfig) => {
*/
const subscribeResponse = (stream$: Observable<WithHttpRequest<HttpEffectResponse>>) =>
stream$
- .pipe(mergeMap(({ request, ...res }) => request.response.send(res)))
+ .pipe(
+ tap(output => pipe(
+ requestMetadataStorage,
+ O.fold(constVoid, metadataStorage => metadataStorage.collect(output)))),
+ mergeMap(({ request, ...res }) => request.response.send(res)))
.subscribe({
error: flow(
unexpectedErrorWhileSendingResponseFactory,

MODIFY
packages/http/src/server/internal-dependencies/httpRequestMetadataStorage.reader.spec.ts
packages/http/src/server/internal-dependencies/httpRequestMetadataStorage.reader.spec.ts
import { createContext } from '@marblejs/core';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
-import { HttpRequestMetadata } from '../../http.interface';
+import { HttpEffectResponse, HttpRequest } from '../..';
+import { HTTP_REQUEST_METADATA_ID_HEADER_KEY } from '../../+internal/metadata.util';
+import { HttpRequestMetadata, WithHttpRequest } from '../../http.interface';
import { HttpRequestMetadataStorage } from './httpRequestMetadataStorage.reader';
describe('#HttpRequestMetadataStorage', () => {
- const storage = HttpRequestMetadataStorage(createContext());
-
test('sets => gets => removes value from storage', () => {
+ // given
const id = createUuid();
+ const storage = HttpRequestMetadataStorage(createContext());
const metadata: HttpRequestMetadata = { test: 'test_data' };
expect(storage.size()).toEqual(0);
+ // when
storage.set(id, metadata);
+ // then
expect(storage.size()).toEqual(1);
+ expect(storage.get(id)).toEqual(metadata);
+ expect(storage.size()).toEqual(0);
+ });
- const retrievedData = storage.get(id);
+ test('collects metadata from request', () => {
+ // given
+ const storage = HttpRequestMetadataStorage(createContext());
+ const meta: HttpRequestMetadata = { test: 'test_data' };
+ const requestId = createUuid();
+ const request = { meta, headers: { [HTTP_REQUEST_METADATA_ID_HEADER_KEY]: requestId } } as any as HttpRequest;
+ const response = { request } as WithHttpRequest<HttpEffectResponse>;
+
+ expect(storage.size()).toEqual(0);
- expect(retrievedData).toEqual(metadata);
+ // when
+ storage.collect(response);
+
+ // then
+ expect(storage.size()).toEqual(1);
+ expect(storage.get(requestId)).toEqual(meta);
expect(storage.size()).toEqual(0);
});
});

MODIFY
packages/http/src/server/internal-dependencies/httpRequestMetadataStorage.reader.ts
packages/http/src/server/internal-dependencies/httpRequestMetadataStorage.reader.ts
import { createContextToken, createReader } from '@marblejs/core';
-import { HttpRequestMetadata } from '../../http.interface';
+import * as O from 'fp-ts/lib/Option';
+import { constVoid, pipe } from 'fp-ts/lib/function';
+import { getHttpRequestMetadataIdHeader } from '../../+internal/metadata.util';
+import { HttpEffectResponse } from '../../effects/http.effects.interface';
+import { HttpRequestMetadata, WithHttpRequest } from '../../http.interface';
-export type HttpRequestMetadataStorage = ReturnType<typeof HttpRequestMetadataStorage>;
+export type HttpRequestMetadataStorage = Readonly<ReturnType<typeof HttpRequestMetadataStorage>>;
export const HttpRequestMetadataStorageToken = createContextToken<HttpRequestMetadataStorage>('HttpRequestMetadataStorage');
export const HttpRequestMetadataStorage = createReader(() => {
const storage = new Map<string, HttpRequestMetadata>();
- const set = (key: string | undefined, metadata: HttpRequestMetadata | undefined) => {
+ const set = (key: string | undefined, metadata: HttpRequestMetadata | undefined): void => {
if (key && metadata) storage.set(key, metadata);
};
- const get = (key: string) => {
+
+ const get = (key: string): HttpRequestMetadata | undefined => {
const metadata = storage.get(key);
storage.delete(key);
return metadata;
@@ -21,9 +26,41 @@ export const HttpRequestMetadataStorage = createReader(() => {
const size = () =>
storage.size;
+ const collect = (response: WithHttpRequest<HttpEffectResponse>): void =>
+ pipe(
+ getHttpRequestMetadataIdHeader(response.request.headers),
+ O.fold(constVoid, requestId => set(requestId, response.request.meta)),
+ );
+
return {
- set,
+ /**
+ * Gets request metadata for given key (request ID) and removes it from storage
+ *
+ * @param key request ID
+ * @returns HttpRequestMetadata | undefined
+ */
get,
+
+ /**
+ * Set request metadata for given key (request ID)
+ *
+ * @param key request ID
+ * @param metadata request metadata
+ * @returns void
+ */
+ set,
+
+ /**
+ * @returns total count of stored request metadatas
+ */
size,
+
+ /**
+ * Collects request metadata from HttpEffect response
+ *
+ * @param response HttpEffectResponse & HttpRequest
+ * @returns void
+ */
+ collect,
};
});

fix(http): collect request metadata in #subscribeResponse