feat(core): global HttpRequestBus
MODIFY
packages/core/src/+internal/testing/http.helper.ts
packages/core/src/+internal/testing/http.helper.ts
import * as http from 'http';
import { EventEmitter } from 'events';
+import { Subject } from 'rxjs';
import {
HttpRequest,
HttpResponse,
@@ -9,9 +10,10 @@ import {
QueryParameters,
HttpServer,
} from '../../http/http.interface';
-import { createContext, lookup } from '../../context/context.factory';
+import { createContext, lookup, registerAll, bindTo } from '../../context/context.factory';
import { createEffectContext } from '../../effects/effectsContext.factory';
import { Server } from '../../http/server/http.server.interface';
+import { HttpRequestBusToken, HttpServerClientToken } from '../../http/server/http.server.tokens';
interface HttpRequestMockParams {
url?: string;
@@ -62,7 +64,11 @@ export const createHttpResponse = (data: HttpResponseMockParams = {}) =>
} as any as HttpResponse;
export const createMockEffectContext = () => {
- const context = createContext();
+ const dependencies = [
+ bindTo(HttpRequestBusToken)(() => new Subject<HttpRequest>()),
+ bindTo(HttpServerClientToken)(() => http.createServer()),
+ ];
+ const context = registerAll(dependencies)(createContext());
const client = http.createServer();
return createEffectContext({ ask: lookup(context), client });
};

MODIFY
packages/core/src/http/response/http.responseHandler.ts
packages/core/src/http/response/http.responseHandler.ts
@@ -2,11 +2,12 @@ import * as url from 'url';
import { EMPTY } from 'rxjs';
import { fold } from 'fp-ts/lib/Option';
import { pipe } from 'fp-ts/lib/pipeable';
+import { constVoid } from 'fp-ts/lib/function';
import { HttpEffectResponse } from '../effects/http.effects.interface';
import { HttpRequest, HttpResponse, HttpStatus } from '../http.interface';
import { isStream } from '../../+internal/utils';
import { ContextProvider } from '../../context/context.factory';
-import { ServerRequestMetadataStorageToken } from '../server/http.server.tokens';
+import { HttpRequestMetadataStorageToken } from '../server/http.server.tokens';
import { getTestingRequestIdHeader, isTestingMetadataOn } from '../../+internal/testing';
import { headersFactory } from './http.responseHeaders.factory';
import { bodyFactory } from './http.responseBody.factory';
@@ -27,8 +28,8 @@ export const handleResponse = (ask: ContextProvider) => (res: HttpResponse) => (
if (isTestingMetadataOn()) {
pipe(
- ask(ServerRequestMetadataStorageToken),
- fold(() => undefined, storage => storage.set(testingHeader, req.meta)),
+ ask(HttpRequestMetadataStorageToken),
+ fold(constVoid, storage => storage.set(testingHeader, req.meta)),
);
}

MODIFY
packages/core/src/http/router/http.router.resolver.v2.ts
packages/core/src/http/router/http.router.resolver.v2.ts
@@ -10,6 +10,8 @@ import {
errorNotBoundToRequestErrorFactory,
responseNotBoundToRequestErrorFactory,
} from '../error/http.error.model';
+import { useContext } from '../../context/context.hook';
+import { HttpRequestBusToken } from '../server/http.server.tokens';
import { Routing, BootstrappedRoutingItem } from './http.router.interface';
import { queryParamsFactory } from './http.router.query.factory';
import { matchRoute } from './http.router.matcher';
@@ -24,6 +26,7 @@ export const resolveRouting = (
output$?: HttpOutputEffect,
error$?: HttpErrorEffect,
) => {
+ const requestBusSubject = useContext(HttpRequestBusToken)(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 }>();
@@ -123,6 +126,7 @@ export const resolveRouting = (
req.meta.path = resolvedRoute.path;
resolvedRoute.subject.next(req);
+ requestBusSubject.next(req);
};
return {

MODIFY
packages/core/src/http/server/http.server.listener.ts
packages/core/src/http/server/http.server.listener.ts
@@ -8,7 +8,7 @@ import { factorizeRoutingWithDefaults } from '../router/http.router.factory';
import { createEffectContext } from '../../effects/effectsContext.factory';
import { useContext } from '../../context/context.hook';
import { createListener } from '../../listener/listener.factory';
-import { ServerClientToken } from './http.server.tokens';
+import { HttpServerClientToken } from './http.server.tokens';
export interface HttpListenerConfig {
effects?: (RouteEffect | RouteEffectGroup)[];
@@ -30,7 +30,7 @@ export const httpListener = createListener<HttpListenerConfig, HttpListener>(con
error$,
} = config ?? {};
- const client = useContext(ServerClientToken)(ask);
+ const client = useContext(HttpServerClientToken)(ask);
const effectContext = createEffectContext({ ask, client });
const routing = factorizeRoutingWithDefaults(effects, middlewares ?? []);
const { resolve } = resolveRouting(routing, effectContext)(output$, error$);

MODIFY
packages/core/src/http/server/http.server.tokens.ts
packages/core/src/http/server/http.server.tokens.ts
-import { Observable } from 'rxjs';
+import { Observable, Subject } from 'rxjs';
import { createContextToken } from '../../context/context.token.factory';
-import { HttpServer } from '../http.interface';
+import { HttpServer, HttpRequest } from '../http.interface';
import { Routing } from '../router/http.router.interface';
import { AllServerEvents } from './http.server.event';
import { ServerRequestMetadataStorage } from './http.server.metadata.storage';
-export const ServerEventStreamToken = createContextToken<Observable<AllServerEvents>>('ServerEventStreamToken');
-export const ServerClientToken = createContextToken<HttpServer>('ServerClientToken');
-export const ServerRoutingToken = createContextToken<Routing>('ServerRoutingToken');
-export const ServerRequestMetadataStorageToken = createContextToken<ServerRequestMetadataStorage>('ServerRequestMetadataStorageToken');
+export const HttpServerEventStreamToken = createContextToken<Observable<AllServerEvents>>('HttpServerEventStreamToken');
+export const HttpServerClientToken = createContextToken<HttpServer>('HttpServerClientToken');
+export const HttpRoutingToken = createContextToken<Routing>('HttpServerRoutingToken');
+export const HttpRequestMetadataStorageToken = createContextToken<ServerRequestMetadataStorage>('HttpServerRequestMetadataStorageToken');
+export const HttpRequestBusToken = createContextToken<Subject<HttpRequest>>('HttpRequestBusToken');

MODIFY
packages/core/src/http/server/http.server.ts
packages/core/src/http/server/http.server.ts
import * as http from 'http';
import * as https from 'https';
import { flow } from 'fp-ts/lib/function';
+import { Subject } from 'rxjs';
import { createContext, lookup, registerAll, bindTo, resolve } from '../../context/context.factory';
import { createEffectContext } from '../../effects/effectsContext.factory';
import { insertIf, isTestingMetadataOn } from '../../+internal';
+import { HttpRequest } from '../http.interface';
import { subscribeServerEvents } from './http.server.event.subscriber';
-import { ServerClientToken, ServerEventStreamToken, ServerRequestMetadataStorageToken } from './http.server.tokens';
+import { HttpServerClientToken, HttpServerEventStreamToken, HttpRequestMetadataStorageToken, HttpRequestBusToken } from './http.server.tokens';
import { serverRequestMetadataStorage } from './http.server.metadata.storage';
import { CreateServerConfig, Server } from './http.server.interface';
@@ -17,15 +19,17 @@ export const createServer = async (config: CreateServerConfig): Promise<Server>
const server = options.httpsOptions ? https.createServer(options.httpsOptions) : http.createServer();
const serverEvent$ = subscribeServerEvents(hostname || DEFAULT_HOSTNAME)(server);
- const boundServerEvent$ = bindTo(ServerEventStreamToken)(() => serverEvent$);
- const boundServer = bindTo(ServerClientToken)(() => server);
- const boundServerRequestMetadataStorage = bindTo(ServerRequestMetadataStorageToken)(serverRequestMetadataStorage);
+ const boundHttpServerEvent = bindTo(HttpServerEventStreamToken)(() => serverEvent$);
+ const boundHttpServerClient = bindTo(HttpServerClientToken)(() => server);
+ const boundHttpRequestMetadataStorage = bindTo(HttpRequestMetadataStorageToken)(serverRequestMetadataStorage);
+ const boundHttpRequestBus = bindTo(HttpRequestBusToken)(() => new Subject<HttpRequest>());
const context = await flow(
registerAll([
- boundServer,
- boundServerEvent$,
- ...insertIf(isTestingMetadataOn(), boundServerRequestMetadataStorage),
+ boundHttpServerClient,
+ boundHttpServerEvent,
+ boundHttpRequestBus,
+ ...insertIf(isTestingMetadataOn(), boundHttpRequestMetadataStorage),
...dependencies,
]),
resolve,

MODIFY
packages/core/src/http/server/specs/http.server.factory.spec.ts
packages/core/src/http/server/specs/http.server.factory.spec.ts
@@ -20,7 +20,7 @@ import {
import { lookup } from '../../../context/context.factory';
import { useContext } from '../../../context/context.hook';
import { HttpServer } from '../../http.interface';
-import { ServerClientToken } from '../http.server.tokens';
+import { HttpServerClientToken } from '../http.server.tokens';
describe('#createServer', () => {
let server: HttpServer;
@@ -93,7 +93,7 @@ describe('#createServer', () => {
server = await app();
const ask = lookup(app.context);
- const boundServer = useContext(ServerClientToken)(ask);
+ const boundServer = useContext(HttpServerClientToken)(ask);
expect(boundServer).toBeDefined();
});

DELETE
packages/core/src/http/server/specs/http.server.listener.spec.ts
-import * as http from 'http';
-import { throwError } from 'rxjs';
-import { mapTo, mergeMap } from 'rxjs/operators';
-import { r } from '../../router/http.router.ixbuilder';
-import { httpListener } from '../http.server.listener';
-import { HttpMiddlewareEffect, HttpEffectResponse } from '../../effects/http.effects.interface';
-import { createContext, registerAll, bindTo, Context } from '../../../context/context.factory';
-import { ServerClientToken } from '../http.server.tokens';
-
-describe('Http listener', () => {
- let responseHandler;
- let context: Context;
-
- beforeAll(() => {
- context = registerAll([
- bindTo(ServerClientToken)(() => http.createServer()),
- ])(createContext());
- });
-
- beforeEach(() => {
- jest.unmock('../../response/http.responseHandler.ts');
- responseHandler = require('../../response/http.responseHandler.ts');
- });
-
- test('#httpListener handles received HttpRequest', done => {
- // given
- const req = { url: '/', method: 'GET' } as http.IncomingMessage;
- const res = {} as http.OutgoingMessage;
- const effectResponse = { body: 'test' } as HttpEffectResponse;
- const sender = jest.fn();
-
- const effect$ = r.pipe(
- r.matchPath('/'),
- r.matchType('GET'),
- r.useEffect(req$ => req$.pipe(
- mapTo(effectResponse),
- )),
- );
-
- const middleware$: HttpMiddlewareEffect = req$ => req$;
-
- // when
- responseHandler.handleResponse = jest.fn(() => () => () => sender);
-
- httpListener({
- middlewares: [middleware$],
- effects: [effect$],
- })(context)(req, res);
-
- // then
- expect(sender).toHaveBeenCalledWith({ ...effectResponse, request: { ...req, response: res }});
- done();
- });
-
- test('#httpListener allows empty middlewares', done => {
- // given
- const req = { url: '/', method: 'GET' } as http.IncomingMessage;
- const res = {} as http.OutgoingMessage;
- const effectResponse = { body: 'test' } as HttpEffectResponse;
- const sender = jest.fn();
-
- const effect$ = r.pipe(
- r.matchPath('/'),
- r.matchType('GET'),
- r.useEffect(req$ => req$.pipe(
- mapTo(effectResponse),
- )),
- );
-
-
- // when
- responseHandler.handleResponse = jest.fn(() => () => () => sender);
-
- httpListener({
- middlewares: [],
- effects: [effect$],
- })(context)(req, res);
-
- // then
- expect(sender).toHaveBeenCalledWith({ body: 'test', request: { ...req, response: res }});
- done();
- });
-
- test('#httpListener catches error', done => {
- // given
- const error = new Error('test');
- const req = { url: '/', method: 'GET' } as http.IncomingMessage;
- const res = {} as http.OutgoingMessage;
- const effectResponse = { body: { error: { message: 'test', status: 500 } }, status: 500 } as HttpEffectResponse;
- const sender = jest.fn();
-
- const effect$ = r.pipe(
- r.matchPath('/'),
- r.matchType('GET'),
- r.useEffect(req$ => req$.pipe(
- mergeMap(() => throwError(error)),
- )),
- );
-
- // when
- responseHandler.handleResponse = jest.fn(() => () => () => sender);
-
- httpListener({
- middlewares: [],
- effects: [effect$],
- })(context)(req, res);
-
- // then
- expect(sender).toHaveBeenCalledWith(effectResponse);
- done();
- });
-});

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
import * as O from 'fp-ts/lib/Option';
import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
-import { reader, ServerEventStreamToken, matchEvent, ServerEvent, AllServerEvents } from '@marblejs/core';
+import { reader, HttpServerEventStreamToken, matchEvent, ServerEvent, AllServerEvents } from '@marblejs/core';
import { from, Observable, EMPTY } from 'rxjs';
import { mergeMap, take, map, mapTo, mergeMapTo } from 'rxjs/operators';
import { TransportMessage, TransportLayerConnection } from '../transport/transport.interface';
@@ -56,7 +56,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
const connection = transportLayer.connect({ isConsumer: false });
pipe(
- ask(ServerEventStreamToken),
+ ask(HttpServerEventStreamToken),
O.map(teardownOnClose$(connection)),
O.getOrElse(() => EMPTY as Observable<any>),
).subscribe();

feat(core): global HttpRequestBus