feat(core): fromReadableStream utility function
MODIFY
packages/core/src/+internal/index.ts
packages/core/src/+internal/index.ts
@@ -3,3 +3,4 @@ export * from './testing';
export * from './utils';
export * from './fp';
export * from './http';
+export * from './observable';

ADD
packages/core/src/+internal/observable/fromReadableStream.ts
+import { Observable } from 'rxjs';
+import { Readable } from 'stream';
+
+export const fromReadableStream = (stream: Readable): Observable<any> => {
+ stream.pause();
+ return new Observable(observer => {
+ const next = chunk => observer.next(chunk);
+ const complete = () => observer.complete();
+ const error = err => observer.error(err);
+
+ stream
+ .on('data', next)
+ .on('error', error)
+ .on('end', complete)
+ .resume();
+
+ return () => {
+ stream.removeListener('data', next);
+ stream.removeListener('error', error);
+ stream.removeListener('end', complete);
+ };
+ });
+};

ADD
packages/core/src/+internal/observable/index.ts
+export * from './fromReadableStream';

feat(core): fromReadableStream utility function