message: refactor(middleware-multipart): busboy event handling
diff: 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))
  ),
  