refactor(http): #responseHandler
MODIFY
packages/core/src/+internal/utils/any.util.ts
packages/core/src/+internal/utils/any.util.ts
@@ -52,13 +52,8 @@ export const pipeFromArray = <T, R>(fns: Array<UnaryFunction<T, R>>): UnaryFunct
};
};
-const fromIO = <T>(fa: IO<T>): Observable<T> =>
+export const fromIO = <T>(fa: IO<T>): Observable<T> =>
defer(() => of(fa()));
-const fromTask = <T>(fa: Task<T>): Observable<T> =>
+export const fromTask = <T>(fa: Task<T>): Observable<T> =>
defer(fa);
-
-export const FpRx = {
- fromIO,
- fromTask,
-};

MODIFY
packages/http/src/+internal/testing.util.ts
packages/http/src/+internal/testing.util.ts
@@ -69,7 +69,7 @@ export const createHttpResponse = (data: HttpResponseMockParams = {}) =>
writeHead = jest.fn();
setHeader = jest.fn();
getHeader = jest.fn();
- end = jest.fn();
+ end = jest.fn((_, cb) => cb());
send = jest.fn(() => of(true));
} as any as HttpResponse;

MODIFY
packages/http/src/http.interface.ts
packages/http/src/http.interface.ts
@@ -29,7 +29,7 @@ export interface QueryParameters {
}
export interface HttpResponse extends http.ServerResponse {
- send: (response: HttpEffectResponse) => Observable<never>;
+ send: (response: HttpEffectResponse) => Observable<boolean>;
}
export interface HttpHeaders extends http.OutgoingHttpHeaders {}

MODIFY
packages/http/src/response/http.responseBody.factory.ts
packages/http/src/response/http.responseBody.factory.ts
@@ -5,9 +5,9 @@ import { HttpHeaders } from '../http.interface';
import { ContentType, getContentTypeUnsafe, isJsonContentType } from '../+internal/contentType.util';
import { transformUrlEncoded } from '../+internal/urlEncoded.util';
-export type ResponseBodyFactory = (headers: HttpHeaders) => (body: any) => string | Stream | Buffer;
+export type ResponseBodyFactory = (data: { headers: HttpHeaders, body: any }) => string | Stream | Buffer;
-export const factorizeBody: ResponseBodyFactory = headers => body => {
+export const factorizeBody: ResponseBodyFactory = ({ headers, body }) => {
const contentType = getContentTypeUnsafe(headers);
if (isStream(body))

MODIFY
packages/http/src/response/http.responseHandler.ts
packages/http/src/response/http.responseHandler.ts
import * as url from 'url';
-import { defer, EMPTY, Observable } from 'rxjs';
-import { ContextProvider } from '@marblejs/core';
-import { isStream } from '@marblejs/core/dist/+internal/utils';
+import { Stream } from 'stream';
+import { Observable } from 'rxjs';
+import { mapTo } from 'rxjs/operators';
+import * as IO from 'fp-ts/lib/IO';
+import * as T from 'fp-ts/lib/Task';
+import { pipe } from 'fp-ts/lib/function';
+import { ContextProvider, Logger, LoggerLevel, LoggerTag, LoggerToken, useContext } from '@marblejs/core';
+import { fromIO, fromTask, isStream } from '@marblejs/core/dist/+internal/utils';
import { HttpEffectResponse } from '../effects/http.effects.interface';
-import { HttpRequest, HttpResponse, HttpStatus } from '../http.interface';
+import { HttpHeaders, HttpRequest, HttpResponse, HttpStatus } from '../http.interface';
import { factorizeHeaders } from './http.responseHeaders.factory';
import { factorizeBody } from './http.responseBody.factory';
@@ -12,29 +17,62 @@ type HandleResponse =
(res: HttpResponse) =>
(req: HttpRequest) =>
(effectResponse: HttpEffectResponse) =>
- Observable<never>;
+ Observable<boolean>;
-const writeBodyAndEndRequest = (res: HttpResponse) => (body: any): Promise<unknown> =>
- new Promise(resolve => res.end(body, () => resolve(undefined)));
+const warnIfOutgoingConnectionEnded = (logger: Logger): IO.IO<void> => () =>
+ logger({
+ tag: LoggerTag.HTTP,
+ level: LoggerLevel.WARN,
+ type: 'Server',
+ message: 'Attempted to send a response to an already finished connection',
+ });
-export const handleResponse: HandleResponse = _ask => res => req => effectResponse => {
- if (res.finished) {
- // @TODO: log that case
- return EMPTY;
- }
+const getResponseStatus = (effectResponse: HttpEffectResponse): HttpStatus =>
+ effectResponse.status ?? HttpStatus.OK;
- const status = effectResponse.status || HttpStatus.OK;
- const path = url.parse(req.url).pathname || '';
+const getRequestUrl = (request: HttpRequest): string =>
+ url.parse(request.url).pathname ?? '';
- const headers = factorizeHeaders({ body: effectResponse.body, path, status })(effectResponse.headers)();
- const body = factorizeBody(headers)(effectResponse.body);
+export const writeHead = (status: HttpStatus, headers: HttpHeaders) => (response: HttpResponse): IO.IO<HttpHeaders> =>
+ () => { response.writeHead(status, headers); return headers; };
- res.writeHead(status, headers);
+export const endRequestAndWriteBody = (body: any) => (res: HttpResponse): T.Task<unknown> =>
+ () => new Promise(resolve => res.end(body, () => resolve(true)));
- if (isStream(body)) {
- body.pipe(res);
- return EMPTY;
- }
+export const endRequest = (res: HttpResponse): T.Task<unknown> =>
+ () => new Promise(resolve => res.end(undefined, () => resolve(true)));
- return defer(() => writeBodyAndEndRequest(res)(body)) as Observable<never>;
+export const streamBody = (body: Stream) => (response: HttpResponse): T.Task<unknown> =>
+ () => (body.pipe(response), Promise.resolve(true));
+
+/**
+ * Send HTTP response
+ *
+ * @sig `ContextProvider -> HttpResponse -> HttpRequest -> HttpEffectResponse -> Observable`
+ * @since 1.0.0
+ */
+export const handleResponse: HandleResponse = ask => {
+ const logger = useContext(LoggerToken)(ask);
+
+ return res => req => effectRes => {
+ if (res.writableEnded)
+ return pipe(
+ warnIfOutgoingConnectionEnded(logger),
+ fromIO,
+ mapTo(false));
+
+ const status = getResponseStatus(effectRes);
+ const path = getRequestUrl(req);
+
+ return pipe(
+ factorizeHeaders({ body: effectRes.body, headers: effectRes.headers, path, status }),
+ IO.chain(headers => writeHead(status, headers)(res)),
+ IO.chain(headers => IO.of(factorizeBody({ headers, body: effectRes.body }))),
+ T.fromIO,
+ T.chain(body => isStream(body)
+ ? streamBody(body)(res)
+ : endRequestAndWriteBody(body)(res)),
+ fromTask,
+ ) as Observable<never>;
+ };
};

MODIFY
packages/http/src/response/http.responseHeaders.factory.ts
packages/http/src/response/http.responseHeaders.factory.ts
@@ -11,6 +11,7 @@ interface HttpResponseLike {
body: any;
path: string;
status: HttpStatus;
+ headers?: HttpHeaders;
}
export const DEFAULT_HEADERS = {
@@ -44,9 +45,9 @@ export const provideContentTypeHeader = (response: HttpResponseLike): HttpHeader
};
};
-export const factorizeHeaders = (response: HttpResponseLike) => (providedHeaders?: HttpHeaders): IO.IO<HttpHeaders> => {
+export const factorizeHeaders = (response: HttpResponseLike): IO.IO<HttpHeaders> => {
const defaultContentTypeHeaders = pipe(
- O.fromNullable(providedHeaders),
+ O.fromNullable(response.headers),
O.chain(getContentType),
O.fold(
() => ({
@@ -54,7 +55,7 @@ export const factorizeHeaders = (response: HttpResponseLike) => (providedHeaders
...provideContentLengthHeader(response),
}),
() => pipe(
- O.fromNullable(providedHeaders),
+ O.fromNullable(response.headers),
O.chain(getContentLength),
O.fold(
() => ({ ...provideContentLengthHeader(response) }),
@@ -63,11 +64,10 @@ export const factorizeHeaders = (response: HttpResponseLike) => (providedHeaders
)),
);
-
const mergedHeaders = {
...DEFAULT_HEADERS,
...defaultContentTypeHeaders,
- ...providedHeaders,
+ ...response.headers,
};
return pipe(

MODIFY
packages/http/src/response/specs/http.responseBody.factory.spec.ts
packages/http/src/response/specs/http.responseBody.factory.spec.ts
@@ -8,7 +8,7 @@ describe('#bodyFactory', () => {
const body = { test: 'test' };
// when
- const factorizedBody = factorizeBody(headers)(body);
+ const factorizedBody = factorizeBody({ headers, body });
// then
expect(factorizedBody).toEqual('test=test');
@@ -20,7 +20,7 @@ describe('#bodyFactory', () => {
const body = { test: 'test' };
// when
- const factorizedBody = factorizeBody(headers)(body);
+ const factorizedBody = factorizeBody({ headers, body });
// then
expect(factorizedBody).toEqual(JSON.stringify(body));
@@ -32,7 +32,7 @@ describe('#bodyFactory', () => {
const bodies = [33, 'test', false];
// when
- const factorizedBodies = bodies.map(factorizeBody(headers));
+ const factorizedBodies = bodies.map(body => factorizeBody({ headers, body }));
// then
expect(factorizedBodies).toEqual(bodies.map(String));
@@ -44,7 +44,7 @@ describe('#bodyFactory', () => {
const body = 'test=test';
// when
- const factorizedBody = factorizeBody(headers)(body);
+ const factorizedBody = factorizeBody({ body, headers });
// then
expect(factorizedBody).toEqual('test=test');
@@ -56,7 +56,7 @@ describe('#bodyFactory', () => {
const body = '1234';
// when
- const factorizedBody = factorizeBody(headers)(body);
+ const factorizedBody = factorizeBody({ headers, body });
// then
expect(factorizedBody).toEqual('1234');

MODIFY
packages/http/src/response/specs/http.responseHandler.spec.ts
packages/http/src/response/specs/http.responseHandler.spec.ts
import * as path from 'path';
import * as fs from 'fs';
-import { Observable } from 'rxjs';
+import { lastValueFrom, Observable } from 'rxjs';
import { HttpResponse, HttpStatus } from '../../http.interface';
import { handleResponse } from '../http.responseHandler';
import { createMockEffectContext, createHttpRequest, createHttpResponse } from '../../+internal/testing.util';
import { ContentType } from '../../+internal/contentType.util';
import { HttpEffectResponse } from '../../effects/http.effects.interface';
-describe('Response handler', () => {
+describe('#handleResponse', () => {
const effectContext = createMockEffectContext();
let request;
let response;
- let handle: (effectResponse: HttpEffectResponse) => Observable<never>;
+ let handle: (effectResponse: HttpEffectResponse) => Observable<boolean>;
beforeEach(() => {
request = createHttpRequest({ url: '/', method: 'GET' });
@@ -20,8 +20,9 @@ describe('Response handler', () => {
handle = handleResponse(effectContext.ask)(response)(request);
});
- test('sets provided status', () => {
- handle({ status: HttpStatus.FORBIDDEN }).subscribe();
+ test('sets provided status', async () => {
+ const output = await lastValueFrom(handle({ status: HttpStatus.FORBIDDEN }));
+ expect(output).toEqual(true);
expect(response.writeHead).toHaveBeenCalledWith(HttpStatus.FORBIDDEN, {
'content-length': 0,
'content-type': 'application/json',
@@ -29,8 +30,9 @@ describe('Response handler', () => {
});
});
- test('sets status OK if is not explicitly set', () => {
- handle({}).subscribe();
+ test('sets status OK if is not explicitly set', async () => {
+ const output = await lastValueFrom(handle({}));
+ expect(output).toEqual(true);
expect(response.writeHead).toHaveBeenCalledWith(HttpStatus.OK, {
'content-length': 0,
'content-type': 'application/json',
@@ -38,8 +40,9 @@ describe('Response handler', () => {
});
});
- test('sets default headers if are not explicitly set', () => {
- handle({}).subscribe();
+ test('sets default headers if are not explicitly set', async () => {
+ const output = await lastValueFrom(handle({}));
+ expect(output).toEqual(true);
expect(response.writeHead).toHaveBeenCalledWith(HttpStatus.OK, {
'content-length': 0,
'content-type': 'application/json',
@@ -47,9 +50,10 @@ describe('Response handler', () => {
});
});
- test('sets provided headers', () => {
+ test('sets provided headers', async () => {
const CUSTOM_HEADERS = { 'Content-Encoding': 'gzip' };
- handle({ headers: CUSTOM_HEADERS, body: { test: 'test' } }).subscribe();
+ const output = await lastValueFrom(handle({ headers: CUSTOM_HEADERS, body: { test: 'test' } }));
+ expect(output).toEqual(true);
expect(response.writeHead).toHaveBeenCalledWith(HttpStatus.OK, {
'content-encoding': 'gzip',
'content-length': 15,
@@ -58,17 +62,19 @@ describe('Response handler', () => {
});
});
- test('sets unefined response if body is not provided', () => {
- handle({}).subscribe();
+ test('sets "undefined" response if body is not provided', async () => {
+ const output = await lastValueFrom(handle({}));
+ expect(output).toEqual(true);
expect(response.end).toHaveBeenCalledWith(undefined, expect.anything());
});
- test('sets stringified response if object body is provided', () => {
- handle({ body: { foo: 'bar' } }).subscribe();
+ test('sets stringified response if object body is provided', async () => {
+ const output = await lastValueFrom(handle({ body: { foo: 'bar' } }));
+ expect(output).toEqual(true);
expect(response.end).toHaveBeenCalledWith(JSON.stringify({ foo: 'bar' }), expect.anything());
});
- test('pipes body stream to response', () => {
+ test('pipes body stream to response', async () => {
// given
const pathToFile = path.resolve(__dirname, '../../../../../assets/media', 'audio.mp3');
const body = fs.createReadStream(pathToFile);
@@ -76,31 +82,23 @@ describe('Response handler', () => {
// when
jest.spyOn(body, 'pipe').mockImplementation(() => jest.fn() as any);
- handle({ body, headers }).subscribe();
+ const output = await lastValueFrom(handle({ body, headers }));
// then
+ expect(output).toEqual(true);
expect(body.pipe).toHaveBeenCalledWith(response);
});
- test('returns immediately if response was finished', done => {
+ test('returns immediately if response was finished', async () => {
// given
- response = { ...createHttpResponse(), finished: true } as HttpResponse;
+ response = { ...createHttpResponse(), writableEnded: true } as HttpResponse;
handle = handleResponse(effectContext.ask)(response)(response);
// when
- const result = handle({});
+ const output = await lastValueFrom(handle({}));
// then
+ expect(output).toEqual(false);
expect(response.end).not.toHaveBeenCalled();
-
- result.subscribe({
- next: () => fail('Stream should be empty'),
- error: () => fail('Stream should be empty'),
- complete: () => {
- expect(true).toBe(true);
- done();
- },
- });
});
-
});

MODIFY
packages/http/src/response/specs/http.responseHeaders.factory.spec.ts
packages/http/src/response/specs/http.responseHeaders.factory.spec.ts
@@ -5,7 +5,7 @@ describe('#factorizeHeaders', () => {
test.each([ null, undefined, '' ])(`returns normalized default headers for "%s" body`, (body: any) => {
const REQUEST = { status: 400, path: '/', body };
- expect(factorizeHeaders(REQUEST)()()).toEqual(normalizeHeaders({
+ expect(factorizeHeaders(REQUEST)()).toEqual(normalizeHeaders({
...DEFAULT_HEADERS,
'content-length': 0,
}));
@@ -14,7 +14,7 @@ describe('#factorizeHeaders', () => {
test('returns normalized default headers if custom are not provided', () => {
const REQUEST = { status: 400, path: '/', body: 'A' };
- expect(factorizeHeaders(REQUEST)()()).toEqual(normalizeHeaders({
+ expect(factorizeHeaders(REQUEST)()).toEqual(normalizeHeaders({
...DEFAULT_HEADERS,
'content-length': 1,
}));
@@ -23,7 +23,7 @@ describe('#factorizeHeaders', () => {
test('returns normalized default headers if custom are not provided for empty object sent as body', () => {
const REQUEST = { status: 400, path: '/', body: {} };
- expect(factorizeHeaders(REQUEST)()()).toEqual(normalizeHeaders({
+ expect(factorizeHeaders(REQUEST)()).toEqual(normalizeHeaders({
...DEFAULT_HEADERS,
'content-length': 2,
}));
@@ -33,7 +33,7 @@ describe('#factorizeHeaders', () => {
const CUSTOM_HEADERS = { 'Content-Encoding': 'gzip' };
const REQUEST = { status: 400, path: '/', body: 'A' };
- expect(factorizeHeaders(REQUEST)(CUSTOM_HEADERS)()).toEqual(normalizeHeaders({
+ expect(factorizeHeaders({ ...REQUEST, headers: CUSTOM_HEADERS })()).toEqual(normalizeHeaders({
...DEFAULT_HEADERS,
...CUSTOM_HEADERS,
'content-length': 1,
@@ -45,7 +45,7 @@ describe('#factorizeHeaders', () => {
const CUSTOM_HEADERS = { 'Content-Encoding': 'gzip' };
const REQUEST = { status: 400, path: '/', body: '123' };
- expect(factorizeHeaders(REQUEST)(CUSTOM_HEADERS)()).toEqual(normalizeHeaders({
+ expect(factorizeHeaders({ ...REQUEST, headers: CUSTOM_HEADERS })()).toEqual(normalizeHeaders({
...DEFAULT_HEADERS,
...CUSTOM_HEADERS,
'content-length': 3,
@@ -56,7 +56,7 @@ describe('#factorizeHeaders', () => {
const CUSTOM_HEADERS = { 'Content-Encoding': 'gzip' };
const REQUEST = { status: 400, path: '/', body: { test: 'TEST' } };
- expect(factorizeHeaders(REQUEST)(CUSTOM_HEADERS)()).toEqual(normalizeHeaders({
+ expect(factorizeHeaders({ ...REQUEST, headers: CUSTOM_HEADERS })()).toEqual(normalizeHeaders({
...DEFAULT_HEADERS,
...CUSTOM_HEADERS,
'content-length': 15,
@@ -67,7 +67,7 @@ describe('#factorizeHeaders', () => {
const CUSTOM_HEADERS = { 'content-type': 'application/json' };
const REQUEST = { status: 400, path: '/', body: 'A' };
- expect(factorizeHeaders(REQUEST)(CUSTOM_HEADERS)()).toEqual(normalizeHeaders({
+ expect(factorizeHeaders({ ...REQUEST, headers: CUSTOM_HEADERS })()).toEqual(normalizeHeaders({
...DEFAULT_HEADERS,
'content-length': 1,
}));

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

MODIFY
packages/middleware-cors/src/middleware.ts
packages/middleware-cors/src/middleware.ts
-import { of, EMPTY } from 'rxjs';
-import { mergeMap } from 'rxjs/operators';
+import { of, EMPTY, defer } from 'rxjs';
+import { mergeMap, mergeMapTo } from 'rxjs/operators';
import { isString } from '@marblejs/core/dist/+internal/utils';
-import { HttpMethod, HttpMiddlewareEffect, HttpStatus } from '@marblejs/http';
+import { endRequest } from '@marblejs/http/dist/response/http.responseHandler';
+import { HttpMethod, HttpMiddlewareEffect, HttpRequest, HttpStatus } from '@marblejs/http';
+import { pipe } from 'fp-ts/lib/pipeable';
import { configurePreflightResponse } from './configurePreflightResponse';
import { configureResponse } from './configureResponse';
@@ -22,26 +24,26 @@ const DEFAULT_OPTIONS: CORSOptions = {
optionsSuccessStatus: HttpStatus.NO_CONTENT,
};
-export const cors$ = (options: CORSOptions = {}): HttpMiddlewareEffect => (req$) => {
+const isCORSRequest = (req: HttpRequest): boolean =>
+ !isString(req.headers.origin);
+
+export const cors$ = (options: CORSOptions = {}): HttpMiddlewareEffect => req$ => {
options = { ...DEFAULT_OPTIONS, ...options };
return req$.pipe(
mergeMap(req => {
- // skip if not a CORS request
- if (!isString(req.headers.origin)) {
+ if (isCORSRequest(req))
return of(req);
- }
if (req.method === 'OPTIONS') {
configurePreflightResponse(req, req.response, options);
- req.response.end();
+ return pipe(
+ defer(endRequest(req.response)),
+ mergeMapTo(EMPTY));
+ }
- return EMPTY;
- } else {
configureResponse(req, req.response, options);
-
return of(req);
- }
}),
);
};

MODIFY
packages/middleware-cors/src/spec/middleware.spec.ts
packages/middleware-cors/src/spec/middleware.spec.ts
import { of } from 'rxjs';
import { Marbles } from '@marblejs/core/dist/+internal/testing';
+import { createHttpRequest } from '@marblejs/http/dist/+internal/testing.util';
import { cors$, CORSOptions } from '../middleware';
import { createMockRequest, createMockEffectContext } from '../util';
describe('CORS middleware', () => {
+ afterEach(jest.clearAllMocks);
+
test('pass through non CORS requests', () => {
// given
const request = createMockRequest('OPTIONS', { origin: null });
@@ -20,9 +23,8 @@ describe('CORS middleware', () => {
});
test('handle CORS preflight request', done => {
- expect.assertions(1);
- const request = createMockRequest('OPTIONS', { origin: 'fake-origin' });
+ const request = createHttpRequest({ method: 'OPTIONS', headers: { origin: 'fake-origin' } });
const ctx = createMockEffectContext();
const request$ = of(request);
@@ -31,7 +33,8 @@ describe('CORS middleware', () => {
middleware$(request$, ctx).subscribe({
next: () => fail('Should not return any further value.'),
complete: () => {
- expect(request.response.setHeader).toBeCalledWith('Access-Control-Allow-Origin', 'fake-origin');
+ expect(request.response.setHeader).toHaveBeenCalledWith('Access-Control-Allow-Origin', 'fake-origin');
+ expect(request.response.setHeader).toHaveBeenCalledWith('Access-Control-Allow-Methods', 'HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS');
done();
},
});

MODIFY
packages/testing/src/testBed/http/http.testBed.ts
packages/testing/src/testBed/http/http.testBed.ts
@@ -47,7 +47,7 @@ const sendRequest = async <T extends HttpMethod>(testBedRequest: HttpTestBedRequ
}
if (testBedRequest.body) {
- const body = factorizeBody(testBedRequest.headers)(testBedRequest.body);
+ const body = factorizeBody({ headers: testBedRequest.headers, body: testBedRequest.body });
return clientRequest.end(body);
}

refactor(http): #responseHandler