fix(middleware-multipart): wait until whole stream is finished
MODIFY
packages/middleware-multipart/src/multipart.parser.file.ts
packages/middleware-multipart/src/multipart.parser.file.ts
@@ -2,16 +2,16 @@ 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, iif } from 'rxjs';
-import { mapTo, map, mergeMap, takeUntil, toArray, tap, mergeMapTo, take, ignoreElements } from 'rxjs/operators';
+import { mapTo, map, mergeMap, takeUntil, toArray, tap, mergeMapTo, ignoreElements } from 'rxjs/operators';
import { Readable } from 'stream';
import { FileIncomingData, ParserOpts } from './multipart.interface';
import { setRequestData, shouldParseFieldname } from './multipart.util';
type FileEvent = [string, NodeJS.ReadableStream, string, string, string];
-const fileSizeLimit = (maxBytes: number | undefined) => (data: FileIncomingData) =>
+const fileSizeLimit = (maxBytes: number | undefined) => (data: FileIncomingData) => (finish$: Observable<any>) =>
fromEvent(data.file, 'limit').pipe(
- take(1),
+ takeUntil(finish$),
mergeMapTo(throwError(
new HttpError(`Reached file size limit for "${data.fieldname}" [${maxBytes} bytes]`, HttpStatus.PRECONDITION_FAILED),
)),
@@ -27,7 +27,7 @@ export const parseFile = (req: HttpRequest) => (opts: ParserOpts) => (event$: Ob
throwError(new HttpError(`File "${data.fieldname}" is not acceptable`, HttpStatus.PRECONDITION_FAILED)),
of(data),
)),
- mergeMap(data => merge(fileSizeLimit(opts.maxFileSize)(data), of(data))),
+ mergeMap(data => merge(fileSizeLimit(opts.maxFileSize)(data)(finish$), of(data))),
mergeMap(data => isNonNullable(opts.stream)
? of(data).pipe(
mergeMap(opts.stream),

MODIFY
packages/middleware-multipart/src/multipart.parser.ts
packages/middleware-multipart/src/multipart.parser.ts
import * as Busboy from 'busboy';
import { HttpRequest, HttpError, HttpStatus } from '@marblejs/core';
import { fromEvent, Observable, of, throwError, merge } from 'rxjs';
-import { mapTo, mergeMap, tap, buffer, mergeMapTo } from 'rxjs/operators';
+import { mapTo, mergeMap, tap, mergeMapTo, takeUntil, last } from 'rxjs/operators';
import { ParserOpts } from './multipart.interface';
import { parseField } from './multipart.parser.field';
import { parseFile } from './multipart.parser.file';
@@ -28,12 +28,14 @@ export const parseMultipart = (opts: ParserOpts) => (req: HttpRequest): Observab
);
const filesLimit$ = fromEvent(busboy, 'filesLimit').pipe(
+ takeUntil(fromEvent(busboy, 'finish')),
mergeMap(() => throwError(
new HttpError(`Reached max files count limit [${opts.maxFileCount}]`, HttpStatus.PRECONDITION_FAILED),
)),
);
const fieldsLimit$ = fromEvent(busboy, 'fieldsLimit').pipe(
+ takeUntil(fromEvent(busboy, 'finish')),
mergeMap(() => throwError(
new HttpError(`Reached max fields count limit [${opts.maxFieldCount}]`, HttpStatus.PRECONDITION_FAILED),
)),
@@ -42,12 +44,12 @@ export const parseMultipart = (opts: ParserOpts) => (req: HttpRequest): Observab
return of(req).pipe(
tap(req => req.pipe(busboy)),
mergeMapTo(merge(
- parseFile$,
- parseField$,
filesLimit$,
fieldsLimit$,
+ parseFile$,
+ parseField$,
)),
- buffer(fromEvent(busboy, 'finish')),
+ last(),
mapTo(req),
);
}

fix(middleware-multipart): wait until whole stream is finished