feat: update the stream service to better use fromEvent

packages/cli/src/ogma.command.ts

@@ -1,7 +1,7 @@
 import { Command, CommandRunner, Option } from 'nest-commander';
 import { Color, OgmaLog } from '@ogma/common';
 import { from, iif, of } from 'rxjs';
-import { filter, map, mergeMap, tap } from 'rxjs/operators';
+import { filter, map, mergeMap, takeUntil, tap } from 'rxjs/operators';
 import { FileService } from './file.service';
 import { badFormat } from './messages';
 import { OgmaGetterService } from './ogma-getters.service';
@@ -43,7 +43,7 @@ export class OgmaCommand implements CommandRunner {
 
   private async runForStdin(options: { color: boolean }): Promise<void> {
     return new Promise((resolve, reject) => {
-      const log$ = this.streamService.readFromStream(process.stdin);
+      const { log: log$, done: done$ } = this.streamService.readFromStream(process.stdin);
       log$
         .pipe(
           mergeMap((val) => {
@@ -52,6 +52,7 @@ export class OgmaCommand implements CommandRunner {
           filter((log) => this.checkOgmaFormat(log)),
           map((jsonLogString) => JSON.parse(jsonLogString)),
           tap((logString) => this.writeLog(logString, options.color)),
+          takeUntil(done$),
         )
         .subscribe({
           error: reject,

packages/cli/src/stream.service.ts

@@ -5,15 +5,13 @@ import { Readable } from 'stream';
 
 @Injectable()
 export class StreamService {
-  readFromStream(stream: Readable): Observable<string> {
-    stream.on('readable', () => {
-      while (null !== stream.read()) {
-        /* no op for data event */
-      }
-    });
-    return fromEvent(stream, 'data').pipe(
-      filter((val) => Buffer.isBuffer(val)),
-      map((val: Buffer) => val.toString('utf-8')),
-    );
+  readFromStream(stream: Readable): { log: Observable<string>; done: Observable<any> } {
+    return {
+      log: fromEvent(stream, 'data').pipe(
+        filter((val) => Buffer.isBuffer(val)),
+        map((val: Buffer) => val.toString('utf-8')),
+      ),
+      done: fromEvent(stream, 'end').pipe(map(() => true)),
+    };
   }
 }