feat(core): added support for node.js streams
ADD
assets/media/audio.mp3
assets/media/audio.mp3
Binary files /dev/null and b/assets/media/audio.mp3 differ

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 { r, combineRoutes, use } from '@marblejs/core';
import { requestValidator$, t } from '@marblejs/middleware-io';
@@ -10,6 +11,17 @@ const getFileValidator$ = requestValidator$({
params: t.type({ dir: t.string })
});
+const getFileStream$ = r.pipe(
+ r.matchPath('/stream/:dir*'),
+ r.matchType('GET'),
+ r.useEffect(req$ => req$.pipe(
+ use(getFileValidator$),
+ map(req => req.params.dir),
+ map(dir => fs.createReadStream(path.resolve(STATIC_PATH, dir))),
+ map(body => ({ body }))
+ )),
+);
+
const getFile$ = r.pipe(
r.matchPath('/:dir*'),
r.matchType('GET'),
@@ -22,5 +34,5 @@ const getFile$ = r.pipe(
export const static$ = combineRoutes(
'/static',
- [ getFile$ ],
+ [ getFileStream$, getFile$ ],
);

MODIFY
packages/core/src/response/response.handler.ts
packages/core/src/response/response.handler.ts
+import * as stream from 'stream';
import { EMPTY } from 'rxjs';
import { HttpEffectResponse } from '../effects/http-effects.interface';
import { HttpRequest, HttpResponse, HttpStatus } from '../http.interface';
@@ -15,12 +16,17 @@ export const handleResponse = (res: HttpResponse) => (req: HttpRequest) => (effe
const bodyFactoryWithHeaders = bodyFactory(headers);
const body = bodyFactoryWithHeaders(effect.body);
+ if (body instanceof stream.Readable) {
+ res.writeHead(status, headers);
+ body.pipe(res);
+ } else {
if (body) {
res.setHeader('Content-Length', Buffer.byteLength(body));
}
res.writeHead(status, headers);
res.end(body);
+ }
return EMPTY;
};

feat(core): added support for node.js streams