feat(rawExec): add custom data listeners support

lib/util/exec/common.spec.ts

@@ -2,6 +2,7 @@ import { spawn as _spawn } from 'node:child_process';
 import type { SendHandle, Serializable } from 'node:child_process';
 import { Readable } from 'node:stream';
 import { mockedFunction, partial } from '../../../test/util';
+import type { DataListener } from './common';
 import { exec } from './common';
 import type { RawExecOptions } from './types';
 
@@ -147,6 +148,10 @@ function getSpawnStub(args: StubArgs): any {
   };
 }
 
+function stringify(list: Buffer[]): string {
+  return Buffer.concat(list).toString('utf8');
+}
+
 describe('util/exec/common', () => {
   const cmd = 'ls -l';
   const stdout = 'out message';
@@ -174,6 +179,42 @@ describe('util/exec/common', () => {
       });
     });
 
+    it('should invoke the output listeners', async () => {
+      const cmd = 'ls -l';
+      const stub = getSpawnStub({
+        cmd,
+        exitCode: 0,
+        exitSignal: null,
+        stdout,
+        stderr,
+      });
+      spawn.mockImplementationOnce((cmd, opts) => stub);
+
+      const stdoutListenerBuffer: Buffer[] = [];
+      const stdoutListener: DataListener = (chunk: Buffer) => {
+        stdoutListenerBuffer.push(chunk);
+      };
+
+      const stderrListenerBuffer: Buffer[] = [];
+      const stderrListener: DataListener = (chunk: Buffer) => {
+        stderrListenerBuffer.push(chunk);
+      };
+
+      await expect(
+        exec(
+          cmd,
+          partial<RawExecOptions>({ encoding: 'utf8', shell: 'bin/bash' }),
+          { stdout: [stdoutListener], stderr: [stderrListener] },
+        ),
+      ).resolves.toEqual({
+        stderr,
+        stdout,
+      });
+
+      expect(stringify(stdoutListenerBuffer)).toEqual(stdout);
+      expect(stringify(stderrListenerBuffer)).toEqual(stderr);
+    });
+
     it('command exits with code 1', async () => {
       const cmd = 'ls -l';
       const stderr = 'err';

lib/util/exec/common.ts

@@ -1,5 +1,7 @@
 import type { ChildProcess } from 'node:child_process';
 import { spawn } from 'node:child_process';
+import type { Readable } from 'node:stream';
+import is from '@sindresorhus/is';
 import type { ExecErrorData } from './exec-error';
 import { ExecError } from './exec-error';
 import type { ExecResult, RawExecOptions } from './types';
@@ -25,15 +27,36 @@ function stringify(list: Buffer[]): string {
   return Buffer.concat(list).toString(encoding);
 }
 
+export type DataListener = (chunk: any) => void;
+
+function registerDataListeners(
+  readable: Readable | null,
+  dataListeners: DataListener[] | undefined,
+): void {
+  if (is.nullOrUndefined(readable) || is.nullOrUndefined(dataListeners)) {
+    return;
+  }
+
+  for (const listener of dataListeners) {
+    readable.on('data', listener);
+  }
+}
+
+type OutputListeners = { stdout?: DataListener[]; stderr?: DataListener[] };
+
 function initStreamListeners(
   cp: ChildProcess,
   opts: RawExecOptions & { maxBuffer: number },
+  outputListeners?: OutputListeners,
 ): [Buffer[], Buffer[]] {
   const stdout: Buffer[] = [];
   const stderr: Buffer[] = [];
   let stdoutLen = 0;
   let stderrLen = 0;
 
+  registerDataListeners(cp.stdout, outputListeners?.stdout);
+  registerDataListeners(cp.stderr, outputListeners?.stderr);
+
   cp.stdout?.on('data', (chunk: Buffer) => {
     // process.stdout.write(data.toString());
     const len = Buffer.byteLength(chunk, encoding);
@@ -58,7 +81,11 @@ function initStreamListeners(
   return [stdout, stderr];
 }
 
-export function exec(cmd: string, opts: RawExecOptions): Promise<ExecResult> {
+export function exec(
+  cmd: string,
+  opts: RawExecOptions,
+  outputListeners?: OutputListeners,
+): Promise<ExecResult> {
   return new Promise((resolve, reject) => {
     const maxBuffer = opts.maxBuffer ?? 10 * 1024 * 1024; // Set default max buffer size to 10MB
     const cp = spawn(cmd, {
@@ -70,10 +97,12 @@ export function exec(cmd: string, opts: RawExecOptions): Promise<ExecResult> {
     });
 
     // handle streams
-    const [stdout, stderr] = initStreamListeners(cp, {
-      ...opts,
-      maxBuffer,
-    });
+    const streamOpts = { ...opts, maxBuffer };
+    const [stdout, stderr] = initStreamListeners(
+      cp,
+      streamOpts,
+      outputListeners,
+    );
 
     // handle process events
     cp.on('error', (error) => {