refactor(middleware-multipart): busboy event handling
MODIFY
packages/middleware-multipart/src/multipart.parser.ts
packages/middleware-multipart/src/multipart.parser.ts
import * as Busboy from 'busboy';
import { HttpRequest, HttpError, HttpStatus, isHttpError } from '@marblejs/core';
-import { fromEvent, Observable, of, throwError, merge } from 'rxjs';
-import { mapTo, mergeMap, tap, mergeMapTo, takeUntil, last, catchError } from 'rxjs/operators';
+import { fromEvent, Observable, throwError, forkJoin } from 'rxjs';
+import { mapTo, mergeMap, takeUntil, catchError, share, defaultIfEmpty } from 'rxjs/operators';
import { ParserOpts } from './multipart.interface';
import { parseField } from './multipart.parser.field';
import { parseFile } from './multipart.parser.file';
@@ -17,42 +17,46 @@ export const parseMultipart = (opts: ParserOpts) => (req: HttpRequest): Observab
},
});
+ const finish$ = fromEvent(busboy, 'finish').pipe(
+ share(),
+ );
+
const parseFile$ = parseFile(req)(opts)(
fromEvent(busboy, 'file'),
- fromEvent(busboy, 'finish'),
+ finish$,
);
const parseField$ = parseField(req)(
fromEvent(busboy, 'field'),
- fromEvent(busboy, 'finish'),
+ finish$,
);
const filesLimit$ = fromEvent(busboy, 'filesLimit').pipe(
- takeUntil(fromEvent(busboy, 'finish')),
+ takeUntil(finish$),
mergeMap(() => throwError(
new HttpError(`Reached max files count limit [${opts.maxFileCount}]`, HttpStatus.PRECONDITION_FAILED, undefined, req),
)),
+ defaultIfEmpty(true),
);
const fieldsLimit$ = fromEvent(busboy, 'fieldsLimit').pipe(
- takeUntil(fromEvent(busboy, 'finish')),
+ takeUntil(finish$),
mergeMap(() => throwError(
new HttpError(`Reached max fields count limit [${opts.maxFieldCount}]`, HttpStatus.PRECONDITION_FAILED, undefined, req),
)),
+ defaultIfEmpty(true),
);
- return of(req).pipe(
- tap(req => req.pipe(busboy)),
- mergeMapTo(merge(
+ req.pipe(busboy);
+
+ return forkJoin(
filesLimit$,
fieldsLimit$,
parseFile$,
parseField$,
- )),
- last(),
+ ).pipe(
mapTo(req),
- catchError((error: Error) =>
- isHttpError(error)
+ catchError((error: Error) => isHttpError(error)
? throwError(error)
: throwError(new HttpError(error.message, HttpStatus.INTERNAL_SERVER_ERROR, undefined, req))
),

refactor(middleware-multipart): busboy event handling