refactor(core): #HttpServerEventStream
DELETE
packages/core/src/http/server/http.server.event.subscriber.ts
-import * as http from 'http';
-import * as net from 'net';
-import { Subject, Observable } from 'rxjs';
-import { HttpServer } from '../http.interface';
-import { ServerEventType, ServerEvent, AllServerEvents } from './http.server.event';
-import { DEFAULT_HOSTNAME } from './http.server.interface';
-
-export const subscribeServerEvents = (hostname?: string) => (httpServer: HttpServer): Observable<AllServerEvents> => {
- const event$ = new Subject<AllServerEvents>();
-
- httpServer.on(ServerEventType.CONNECT, (req: http.IncomingMessage, socket: net.Socket, head: Buffer) =>
- event$.next(ServerEvent.connect(req, socket, head)),
- );
-
- httpServer.on(ServerEventType.CONNECTION, (socket: net.Socket) =>
- event$.next(ServerEvent.connection(socket)),
- );
-
- httpServer.on(ServerEventType.CLIENT_ERROR, (error: Error, socket: net.Socket) =>
- event$.next(ServerEvent.clientError(error, socket)),
- );
-
- httpServer.on(ServerEventType.CLOSE, () =>
- event$.next(ServerEvent.close()),
- );
-
- httpServer.on(ServerEventType.CHECK_CONTINUE, (req: http.IncomingMessage, res: http.ServerResponse) =>
- event$.next(ServerEvent.checkContinue(req, res)),
- );
-
- httpServer.on(ServerEventType.CHECK_EXPECTATION, (req: http.IncomingMessage, res: http.ServerResponse) =>
- event$.next(ServerEvent.checkExpectation(req, res)),
- );
-
- httpServer.on(ServerEventType.ERROR, (error: Error) =>
- event$.next(ServerEvent.error(error)),
- );
-
- httpServer.on(ServerEventType.REQUEST, (req: http.IncomingMessage, res: http.ServerResponse) =>
- event$.next(ServerEvent.request(req, res)),
- );
-
- httpServer.on(ServerEventType.UPGRADE, (req: http.IncomingMessage, socket: net.Socket, head: Buffer) =>
- event$.next(ServerEvent.upgrade(req, socket, head)),
- );
-
- httpServer.on(ServerEventType.LISTENING, () => {
- const serverAddressInfo = httpServer.address() as net.AddressInfo;
- event$.next(ServerEvent.listening(serverAddressInfo.port, hostname ?? DEFAULT_HOSTNAME));
- });
-
- return event$.asObservable();
-};

RENAME
packages/core/src/http/server/specs/http.server.factory.spec.ts
packages/core/src/http/server/http.server.spec.ts
@@ -4,8 +4,11 @@ import * as https from 'https';
import { EventEmitter } from 'events';
import { forkJoin } from 'rxjs';
import { tap, filter, take } from 'rxjs/operators';
-import { httpListener } from '../http.server.listener';
-import { createServer } from '../http.server';
+import { HttpServer } from '../http.interface';
+import { lookup } from '../../context/context.factory';
+import { useContext } from '../../context/context.hook';
+import { httpListener } from './http.server.listener';
+import { createServer } from './http.server';
import {
ServerEventType,
isListeningEvent,
@@ -17,11 +20,8 @@ import {
isCheckExpectationEvent,
isClientErrorEvent,
isCloseEvent,
-} from '../http.server.event';
-import { lookup } from '../../../context/context.factory';
-import { useContext } from '../../../context/context.hook';
-import { HttpServer } from '../../http.interface';
-import { HttpServerClientToken } from '../http.server.tokens';
+} from './http.server.event';
+import { HttpServerClientToken } from './http.server.tokens';
describe('#createServer', () => {
let server: HttpServer;
@@ -69,8 +69,8 @@ describe('#createServer', () => {
test('creates https server', async () => {
// given
const httpsOptions: https.ServerOptions = {
- key: fs.readFileSync(path.resolve(__dirname, '../../../../../../assets/key.pem')),
- cert: fs.readFileSync(path.resolve(__dirname, '../../../../../../assets/cert.pem')),
+ key: fs.readFileSync(path.resolve(__dirname, '../../../../../assets/key.pem')),
+ cert: fs.readFileSync(path.resolve(__dirname, '../../../../../assets/cert.pem')),
};
const app = await createServer({
listener: httpListener(),

MODIFY
packages/core/src/http/server/http.server.tokens.ts
packages/core/src/http/server/http.server.tokens.ts
-import { Observable, Subject } from 'rxjs';
+import { Subject } from 'rxjs';
import { createContextToken } from '../../context/context.token.factory';
import { HttpServer, HttpRequest } from '../http.interface';
-import { AllServerEvents } from './http.server.event';
-export const HttpServerEventStreamToken = createContextToken<Observable<AllServerEvents>>('HttpServerEventStreamToken');
export const HttpServerClientToken = createContextToken<HttpServer>('HttpServerClientToken');
export const HttpRequestBusToken = createContextToken<Subject<HttpRequest>>('HttpRequestBusToken');

MODIFY
packages/core/src/http/server/http.server.ts
packages/core/src/http/server/http.server.ts
@@ -2,7 +2,8 @@ import * as http from 'http';
import * as https from 'https';
import { Subject, merge, EMPTY } from 'rxjs';
import { takeWhile } from 'rxjs/operators';
-import { lookup, bindTo } from '../../context/context.factory';
+import { pipe } from 'fp-ts/lib/pipeable';
+import { lookup, bindTo, bindEagerlyTo } from '../../context/context.factory';
import { createEffectContext } from '../../effects/effectsContext.factory';
import { isTestingMetadataOn } from '../../+internal/testing';
import { insertIf } from '../../+internal/utils';
@@ -12,19 +13,23 @@ import { contextFactory } from '../../context/context.factory.helper';
import { ServerIO } from '../../listener/listener.interface';
import { listening$, close$, error$ } from '../effects/http.effects';
import { LoggerTag } from '../../logger';
-import { subscribeServerEvents } from './http.server.event.subscriber';
-import { HttpServerClientToken, HttpServerEventStreamToken, HttpRequestBusToken } from './http.server.tokens';
-import { HttpRequestMetadataStorage, HttpRequestMetadataStorageToken } from './internal-dependencies/httpRequestMetadataStorage.reader';
+import { useContext } from '../../context/context.hook';
+import { HttpServerClientToken, HttpRequestBusToken } from './http.server.tokens';
import { CreateServerConfig } from './http.server.interface';
import { isCloseEvent } from './http.server.event';
+// internal dependencies
+import { HttpRequestMetadataStorage, HttpRequestMetadataStorageToken } from './internal-dependencies/httpRequestMetadataStorage.reader';
+import { HttpServerEventStreamToken, HttpServerEventStream } from './internal-dependencies/httpServerEventStream.reader';
+
export const createServer = async (config: CreateServerConfig) => {
const { listener, event$, port, hostname, dependencies = [], options = {} } = config;
- const server = options.httpsOptions ? https.createServer(options.httpsOptions) : http.createServer();
- const serverEvent$ = subscribeServerEvents(hostname)(server);
+ const server = options.httpsOptions
+ ? https.createServer(options.httpsOptions)
+ : http.createServer();
- const boundHttpServerEvent = bindTo(HttpServerEventStreamToken)(() => serverEvent$);
+ const boundHttpServerEvent = bindEagerlyTo(HttpServerEventStreamToken)(HttpServerEventStream({ server, hostname }));
const boundHttpServerClient = bindTo(HttpServerClientToken)(() => server);
const boundHttpRequestMetadataStorage = bindTo(HttpRequestMetadataStorageToken)(HttpRequestMetadataStorage);
const boundHttpRequestBus = bindTo(HttpRequestBusToken)(() => new Subject<HttpRequest>());
@@ -42,14 +47,15 @@ export const createServer = async (config: CreateServerConfig) => {
const ask = lookup(context);
const ctx = createEffectContext({ ask, client: server });
const httpListener = listener(context);
+ const serverEvent$ = useContext(HttpServerEventStreamToken)(ask);
+ pipe(
merge(
event$ ? event$(serverEvent$, ctx) : EMPTY,
listening$(serverEvent$, ctx),
error$(serverEvent$, ctx),
- close$(serverEvent$, ctx),
- ).pipe(
- takeWhile(e => !isCloseEvent(e), true)
+ close$(serverEvent$, ctx)),
+ takeWhile(e => !isCloseEvent(e), true),
).subscribe();
const listen: ServerIO<HttpServer> = () => new Promise((resolve, reject) => {

ADD
packages/core/src/http/server/internal-dependencies/httpServerEventStream.reader.spec.ts
+import * as http from 'http';
+import { EventEmitter } from 'events';
+import { forkJoin } from 'rxjs';
+import { filter, take } from 'rxjs/operators';
+import { createContext } from '../../../context/context.factory';
+import {
+ isErrorEvent,
+ isClientErrorEvent,
+ isCloseEvent,
+ isConnectEvent,
+ isConnectionEvent,
+ isListeningEvent,
+ isUpgradeEvent,
+ isCheckContinueEvent,
+ isCheckExpectationEvent,
+ ServerEventType,
+} from '../../server/http.server.event';
+import { closeServer } from '../../../+internal/utils';
+import { HttpServerEventStream } from './httpServerEventStream.reader';
+
+describe('#HttpServerEventStream', () => {
+ let server: http.Server;
+
+ beforeEach(() => {
+ server = http.createServer().listen();
+ });
+
+ afterEach(async () => {
+ await closeServer(server)();
+ });
+
+ test('emits mapped event types', async () => {
+ // given
+ const hostname = '127.0.0.1';
+ const event$ = HttpServerEventStream({ hostname, server })(createContext());
+
+ // when
+ const result = forkJoin(
+ event$.pipe(filter(isErrorEvent), take(1)),
+ event$.pipe(filter(isClientErrorEvent), take(1)),
+ event$.pipe(filter(isCloseEvent), take(1)),
+ event$.pipe(filter(isConnectEvent), take(1)),
+ event$.pipe(filter(isConnectionEvent), take(1)),
+ event$.pipe(filter(isListeningEvent), take(1)),
+ event$.pipe(filter(isUpgradeEvent), take(1)),
+ event$.pipe(filter(isCheckContinueEvent), take(1)),
+ event$.pipe(filter(isCheckExpectationEvent), take(1)),
+ ).toPromise();
+
+ server.emit(ServerEventType.ERROR, new Error('test_error'));
+ server.emit(ServerEventType.CLIENT_ERROR);
+ server.emit(ServerEventType.CLOSE);
+ server.emit(ServerEventType.CONNECT);
+ server.emit(ServerEventType.CONNECTION, new EventEmitter());
+ server.emit(ServerEventType.LISTENING);
+ server.emit(ServerEventType.UPGRADE);
+ server.emit(ServerEventType.CHECK_CONTINUE);
+ server.emit(ServerEventType.CHECK_EXPECTATION);
+
+ // then
+ expect(result).resolves.toHaveLength(9);
+ });
+});

ADD
packages/core/src/http/server/internal-dependencies/httpServerEventStream.reader.ts
+import * as http from 'http';
+import * as net from 'net';
+import { Subject } from 'rxjs';
+import { createReader } from '../../../context/context.reader.factory';
+import { HttpServer } from '../../http.interface';
+import { AllServerEvents, ServerEventType, ServerEvent } from '../http.server.event';
+import { DEFAULT_HOSTNAME } from '../http.server.interface';
+import { createContextToken } from '../../../context/context.token.factory';
+
+type HttpServerEventStreamOptions = { server: HttpServer, hostname?: string };
+
+type HttpServerEventStream = ReturnType<ReturnType<typeof HttpServerEventStream>>;
+
+export const HttpServerEventStreamToken = createContextToken<HttpServerEventStream>('HttpServerEventStream');
+
+/**
+ * HTTP server event stream reader responsible for subscribing to raw Node.js HTTP server events
+ * and giving them in normalized Observable form
+ */
+export const HttpServerEventStream = (opts: HttpServerEventStreamOptions) => createReader(_ => {
+ const { server, hostname } = opts;
+ const event$ = new Subject<AllServerEvents>();
+
+ server.on(ServerEventType.CONNECT, (req: http.IncomingMessage, socket: net.Socket, head: Buffer) =>
+ event$.next(ServerEvent.connect(req, socket, head)),
+ );
+
+ server.on(ServerEventType.CONNECTION, (socket: net.Socket) =>
+ event$.next(ServerEvent.connection(socket)),
+ );
+
+ server.on(ServerEventType.CLIENT_ERROR, (error: Error, socket: net.Socket) =>
+ event$.next(ServerEvent.clientError(error, socket)),
+ );
+
+ server.on(ServerEventType.CLOSE, () =>
+ event$.next(ServerEvent.close()),
+ );
+
+ server.on(ServerEventType.CHECK_CONTINUE, (req: http.IncomingMessage, res: http.ServerResponse) =>
+ event$.next(ServerEvent.checkContinue(req, res)),
+ );
+
+ server.on(ServerEventType.CHECK_EXPECTATION, (req: http.IncomingMessage, res: http.ServerResponse) =>
+ event$.next(ServerEvent.checkExpectation(req, res)),
+ );
+
+ server.on(ServerEventType.ERROR, (error: Error) =>
+ event$.next(ServerEvent.error(error)),
+ );
+
+ server.on(ServerEventType.REQUEST, (req: http.IncomingMessage, res: http.ServerResponse) =>
+ event$.next(ServerEvent.request(req, res)),
+ );
+
+ server.on(ServerEventType.UPGRADE, (req: http.IncomingMessage, socket: net.Socket, head: Buffer) =>
+ event$.next(ServerEvent.upgrade(req, socket, head)),
+ );
+
+ server.on(ServerEventType.LISTENING, () => {
+ const serverAddressInfo = server.address() as net.AddressInfo;
+ event$.next(ServerEvent.listening(serverAddressInfo.port, hostname ?? DEFAULT_HOSTNAME));
+ });
+
+ return event$.asObservable();
+});

MODIFY
packages/core/src/index.spec.ts
packages/core/src/index.spec.ts
@@ -28,5 +28,9 @@ describe('@marblejs/core public API', () => {
expect(API.act).toBeDefined();
expect(API.switchToProtocol).toBeDefined();
expect(API.matchEvent).toBeDefined();
+
+ // internal dependencies
+ expect(API.HttpServerEventStreamToken).toBeDefined();
+ expect(API.HttpRequestMetadataStorageToken).toBeDefined();
});
});

MODIFY
packages/core/src/index.ts
packages/core/src/index.ts
@@ -15,6 +15,7 @@ export * from './http/http.interface';
// http - server - internal dependencies
export * from './http/server/internal-dependencies/httpRequestMetadataStorage.reader';
+export * from './http/server/internal-dependencies/httpServerEventStream.reader';
// core - error
export { coreErrorFactory, CoreErrorOptions } from './error/error.factory';

refactor(core): #HttpServerEventStream