feat(middleware-multipart): multipart/form-data handling middleware
MODIFY
.gitignore
.gitignore
@@ -25,6 +25,7 @@ build
dist
doc
coverage
+tmp
package-lock.json
*.tgz

MODIFY
packages/@integration/package.json
packages/@integration/package.json
"@marblejs/middleware-body": "^2.2.2",
"@marblejs/middleware-cors": "^2.2.2",
"@marblejs/middleware-io": "^2.2.2",
+ "@marblejs/middleware-multipart": "^2.2.2",
"@marblejs/middleware-logger": "^2.2.2",
"@marblejs/websockets": "^2.2.2",
"@types/node": "^10.1.2",

MODIFY
packages/@integration/src/effects/static.effects.ts
packages/@integration/src/effects/static.effects.ts
import * as fs from 'fs';
import * as path from 'path';
+import { promisify } from 'util';
import { r, combineRoutes, use } from '@marblejs/core';
import { requestValidator$, t } from '@marblejs/middleware-io';
+import { multipart$, StreamHandler } from '@marblejs/middleware-multipart';
import { readFile } from '@marblejs/core/dist/+internal';
import { map, mergeMap } from 'rxjs/operators';
+const fsExists = promisify(fs.exists);
+const fsMakeDir = promisify(fs.mkdir);
+
const STATIC_PATH = path.resolve(__dirname, '../../../../assets');
+const TMP_PATH = path.resolve(__dirname, '../../../../tmp');
+
+const streamFileTo: StreamHandler = ({ file, fieldname }) =>
+ fsExists(TMP_PATH)
+ .then(exists => !exists
+ ? fsMakeDir(TMP_PATH)
+ : Promise.resolve())
+ .then(() => {
+ const destination = path.join(TMP_PATH, path.basename(fieldname));
+ file.pipe(fs.createWriteStream(destination));
+ return { destination };
+ });
const getFileValidator$ = requestValidator$({
params: t.type({ dir: t.string })
});
+const postFile$ = r.pipe(
+ r.matchPath('/upload'),
+ r.matchType('POST'),
+ r.useEffect(req$ => req$.pipe(
+ use(multipart$({ stream: streamFileTo })),
+ map(req => ({ body: req.file }))
+ )));
+
const getFileStream$ = r.pipe(
r.matchPath('/stream/:dir*'),
r.matchType('GET'),
@@ -18,7 +43,7 @@ const getFileStream$ = r.pipe(
use(getFileValidator$),
map(req => req.params.dir),
map(dir => fs.createReadStream(path.resolve(STATIC_PATH, dir))),
- map(body => ({ body }))
+ map(body => ({ body })),
)),
);
@@ -34,5 +59,5 @@ const getFile$ = r.pipe(
export const static$ = combineRoutes(
'/static',
- [ getFileStream$, getFile$ ],
+ [ postFile$, getFileStream$, getFile$ ],
);

ADD
packages/middleware-multipart/README.md
+<p align="center">
+ <a href="https://marblejs.com">
+ <img src="https://github.com/marblejs/marble/blob/master/assets/img/logo.png?raw=true" width="200" alt="Marble.js logo"/>
+ </a>
+</p>
+
+# @marblejs/middleware-multipart
+
+A multipart/form-data middleware for [Marble.js](https://github.com/marblejs/marble).
+
+## Installation
+
+```
+$ npm i @marblejs/middleware-multipart
+```
+Requires `@marblejs/core` to be installed.
+
+## Documentation
+
+For the latest updates, documentation, change log, and release information visit [docs.marblejs.com](https://docs.marblejs.com) and follow [@marble_js](https://twitter.com/marble_js) on Twitter.
+
+License: MIT

ADD
packages/middleware-multipart/package.json
+{
+ "name": "@marblejs/middleware-multipart",
+ "version": "2.2.2",
+ "description": "A multipart/form-data middleware for Marble.js",
+ "main": "dist/index.js",
+ "types": "./dist/index.d.ts",
+ "scripts": {
+ "start": "yarn watch",
+ "watch": "tsc -w",
+ "build": "tsc",
+ "test": "jest --config ../../jest.config.js",
+ "clean": "rimraf dist"
+ },
+ "files": [
+ "dist/"
+ ],
+ "keywords": [
+ "marble.js",
+ "multipart",
+ "middleware",
+ "http",
+ "rxjs"
+ ],
+ "repository": {
+ "type": "git",
+ "url": "git+https://github.com/marblejs/marble.git"
+ },
+ "engines": {
+ "node": ">= 8.0.0",
+ "yarn": ">= 1.7.0",
+ "npm": ">= 5.0.0"
+ },
+ "author": "marblejs",
+ "license": "MIT",
+ "bugs": {
+ "url": "https://github.com/marblejs/marble/issues"
+ },
+ "homepage": "https://github.com/marblejs/marble#readme",
+ "dependencies": {
+ "busboy": "~0.3.1",
+ "@types/busboy": "^0.2.3"
+ },
+ "peerDependencies": {
+ "@marblejs/core": "^2.1.0",
+ "rxjs": "^6.3.3"
+ },
+ "devDependencies": {
+ "@marblejs/core": "^2.2.2"
+ },
+ "publishConfig": {
+ "access": "public"
+ }
+}

ADD
packages/middleware-multipart/src/index.ts
+export * from './multipart.middleware';
+export { StreamHandler } from './multipart.parser';

ADD
packages/middleware-multipart/src/multipart.interface.ts
+export interface File {
+ destination?: any;
+ buffer?: Buffer;
+ size?: number;
+ encoding: string;
+ mimetype: string;
+ filename: string;
+ fieldname: string;
+}
+
+export interface WithFile {
+ file: Record<string, File>;
+}

ADD
packages/middleware-multipart/src/multipart.middleware.ts
+import { HttpRequest } from '@marblejs/core';
+import { of, Observable } from 'rxjs';
+import { mergeMap, map } from 'rxjs/operators';
+import { parseMultipart, ParserOpts } from './multipart.parser';
+import { shouldParseMultipart } from './multipart.util';
+import { WithFile } from './multipart.interface';
+
+export const multipart$ = (opts: ParserOpts = {}) => <T extends HttpRequest>(req$: Observable<T>) =>
+ req$.pipe(
+ mergeMap(req => shouldParseMultipart(req)
+ ? parseMultipart(opts)(req)
+ : of(req)),
+ map(req => req as T & WithFile),
+ );

ADD
packages/middleware-multipart/src/multipart.parser.ts
+import { HttpRequest, HttpError, HttpStatus } from '@marblejs/core';
+import { isNonNullable } from '@marblejs/core/dist/+internal/utils';
+import { fromReadableStream } from '@marblejs/core/dist/+internal/observable';
+import { fromEvent, Observable, of, throwError, merge } from 'rxjs';
+import { mapTo, map, mergeMap, takeUntil, toArray, tap, catchError, buffer, mergeMapTo, take, ignoreElements } from 'rxjs/operators';
+import * as Busboy from 'busboy';
+import { Readable } from 'stream';
+
+type FileEvent = [string, NodeJS.ReadableStream, string, string, string];
+
+type FieldEvent = [string, any, boolean, boolean, string, string];
+
+type FileData = {
+ file: NodeJS.ReadableStream;
+ filename: string;
+ fieldname: string;
+ encoding: string;
+ mimetype: string;
+};
+
+export interface StreamHandler {
+ (opts: FileData): Promise<{ destination: any }> | Observable<{ destination: any }>;
+}
+
+export interface ParserOpts {
+ stream?: StreamHandler;
+ maxFileSize?: number;
+ maxFileCount?: number;
+}
+
+const fileSizeLimit = (data: FileData, maxBytes: number | undefined) =>
+ fromEvent(data.file, 'limit').pipe(
+ take(1),
+ mergeMapTo(throwError(
+ new HttpError(`Reached file size limit for "${data.fieldname}" [${maxBytes} bytes]`, HttpStatus.PRECONDITION_FAILED),
+ )),
+ ignoreElements(),
+ );
+
+const parseFile = (req: HttpRequest) => (opts: ParserOpts) => (event$: Observable<FileEvent>, finish$: Observable<any>) =>
+ event$.pipe(
+ takeUntil(finish$),
+ map(([ fieldname, file, filename, encoding, mimetype ]) => ({ fieldname, file, filename, encoding, mimetype })),
+ mergeMap((data) => merge(fileSizeLimit(data, opts.maxFileSize), of(data))),
+ mergeMap(data => isNonNullable(opts.stream)
+ ? of(data).pipe(
+ mergeMap(opts.stream),
+ catchError(error => throwError(new HttpError(error.message, HttpStatus.INTERNAL_SERVER_ERROR))),
+ tap(({ destination }) => {
+ req.file = {
+ ...req.file || {},
+ [data.fieldname]: {
+ destination,
+ encoding: data.encoding,
+ mimetype: data.mimetype,
+ filename: data.filename,
+ fieldname: data.fieldname,
+ },
+ };
+ }),
+ mapTo(data),
+ )
+ : of(data.file as Readable).pipe(
+ mergeMap(fromReadableStream),
+ toArray(),
+ map(chunks => Buffer.concat(chunks)),
+ catchError(error => throwError(new HttpError(error.message, HttpStatus.INTERNAL_SERVER_ERROR))),
+ tap(buffer => {
+ req.file = {
+ ...req.file || {},
+ [data.fieldname]: {
+ buffer,
+ encoding: data.encoding,
+ mimetype: data.mimetype,
+ filename: data.filename,
+ fieldname: data.fieldname,
+ size: Buffer.byteLength(buffer),
+ },
+ };
+ }),
+ mapTo(data),
+ ),
+ ),
+ );
+
+const parseField = (req: HttpRequest) => (event$: Observable<FieldEvent>, finish$: Observable<any>) =>
+ event$.pipe(
+ takeUntil(finish$),
+ tap(([ fieldname, value ]) => {
+ req.body = {
+ ...req.body || {},
+ [fieldname]: value,
+ };
+ }),
+ );
+
+export const parseMultipart = (opts: ParserOpts = {}) => (req: HttpRequest): Observable<HttpRequest> => {
+ const busboy = new Busboy({
+ headers: req.headers,
+ limits: {
+ fileSize: opts.maxFileSize,
+ files: opts.maxFileCount,
+ },
+ });
+
+ const parseFile$ = parseFile(req)(opts)(
+ fromEvent(busboy, 'file'),
+ fromEvent(busboy, 'finish'),
+ );
+
+ const parseField$ = parseField(req)(
+ fromEvent(busboy, 'field'),
+ fromEvent(busboy, 'finish'),
+ );
+
+ const filesLimit$ = fromEvent(busboy, 'filesLimit').pipe(
+ mergeMap(() => throwError(
+ new HttpError(`Reached max files count limit [${opts.maxFileCount}]`, HttpStatus.PRECONDITION_FAILED),
+ )),
+ );
+
+ return of(req).pipe(
+ tap(() => req.pipe(busboy)),
+ mergeMapTo(merge(parseFile$, parseField$, filesLimit$)),
+ buffer(fromEvent(busboy, 'finish')),
+ mapTo(req),
+ );
+}

ADD
packages/middleware-multipart/src/multipart.util.ts
+import { HttpRequest } from '@marblejs/core';
+import { getContentType } from '@marblejs/core/dist/+internal/http';
+
+const isProperMethod = (req: HttpRequest): boolean =>
+ ['POST', 'PUT'].includes(req.method);
+
+const isMultipart = (req: HttpRequest): boolean =>
+ getContentType(req.headers).includes('multipart/');
+
+export const shouldParseMultipart = (req: HttpRequest) =>
+ isProperMethod(req) && isMultipart(req);

ADD
packages/middleware-multipart/tsconfig.json
+{
+ "extends": "../../tsconfig.json",
+ "compilerOptions": {
+ "outDir": "./dist/",
+ },
+ "include": ["src/**/*.ts"],
+}

MODIFY
yarn.lock
yarn.lock
resolved "https://registry.yarnpkg.com/@types/aws-lambda/-/aws-lambda-8.10.26.tgz#ebb164a95317b1350c5a57f719e8cb803a76fc97"
integrity sha512-RI32vdCHAtKaFpjA9Wp2+VGrE4aagsiHhAEopWCcMDZreZjyC26Wlnwu2CZvqycMjKjZ1tLhpHIJC/VmjJIspQ==
+"@types/busboy@^0.2.3":
+ version "0.2.3"
+ resolved "https://registry.yarnpkg.com/@types/busboy/-/busboy-0.2.3.tgz#6697ad29873246c530f09a3ff5a40861824230d5"
+ integrity sha1-ZpetKYcyRsUw8Jo/9aQIYYJCMNU=
+ dependencies:
+ "@types/node" "*"
+
"@types/chalk@^2.2.0":
version "2.2.0"
resolved "https://registry.yarnpkg.com/@types/chalk/-/chalk-2.2.0.tgz#b7f6e446f4511029ee8e3f43075fb5b73fbaa0ba"
@@ -1310,6 +1317,13 @@ builtins@^1.0.3:
version "1.0.3"
resolved "https://registry.yarnpkg.com/builtins/-/builtins-1.0.3.tgz#cb94faeb61c8696451db36534e1422f94f0aee88"
+busboy@~0.3.1:
+ version "0.3.1"
+ resolved "https://registry.yarnpkg.com/busboy/-/busboy-0.3.1.tgz#170899274c5bf38aae27d5c62b71268cd585fd1b"
+ integrity sha512-y7tTxhGKXcyBxRKAni+awqx8uqaJKrSFSNFSeRG5CsWNdmy2BIK+6VGWEW7TZnIO/533mtMEA4rOevQV815YJw==
+ dependencies:
+ dicer "0.3.0"
+
byline@^5.0.0:
version "5.0.0"
resolved "https://registry.yarnpkg.com/byline/-/byline-5.0.0.tgz#741c5216468eadc457b03410118ad77de8c1ddb1"
@@ -1977,6 +1991,13 @@ dezalgo@^1.0.0:
asap "^2.0.0"
wrappy "1"
+dicer@0.3.0:
+ version "0.3.0"
+ resolved "https://registry.yarnpkg.com/dicer/-/dicer-0.3.0.tgz#eacd98b3bfbf92e8ab5c2fdb71aaac44bb06b872"
+ integrity sha512-MdceRRWqltEG2dZqO769g27N/3PXfcKl04VhYnBlo2YhH7zPi88VebsjTKclaOyiuMaGU72hTfw3VkUitGcVCA==
+ dependencies:
+ streamsearch "0.1.2"
+
diff@^3.1.0, diff@^3.2.0:
version "3.5.0"
resolved "https://registry.yarnpkg.com/diff/-/diff-3.5.0.tgz#800c0dd1e0a8bfbc95835c202ad220fe317e5a12"
@@ -6043,6 +6064,11 @@ stream-shift@^1.0.0:
version "1.0.0"
resolved "https://registry.yarnpkg.com/stream-shift/-/stream-shift-1.0.0.tgz#d5c752825e5367e786f78e18e445ea223a155952"
+streamsearch@0.1.2:
+ version "0.1.2"
+ resolved "https://registry.yarnpkg.com/streamsearch/-/streamsearch-0.1.2.tgz#808b9d0e56fc273d809ba57338e929919a1a9f1a"
+ integrity sha1-gIudDlb8Jz2Am6VzOOkpkZoanxo=
+
string-argv@^0.0.2:
version "0.0.2"
resolved "https://registry.yarnpkg.com/string-argv/-/string-argv-0.0.2.tgz#dac30408690c21f3c3630a3ff3a05877bdcbd736"

feat(middleware-multipart): multipart/form-data handling middleware