feat(rawExec): add custom data listeners support (#34066)

Co-authored-by: Rhys Arkins <[email protected]>

lib/util/exec/common.spec.ts

@@ -3,7 +3,7 @@ import type { SendHandle, Serializable } from 'node:child_process';
 import { Readable } from 'node:stream';
 import { mockedFunction, partial } from '../../../test/util';
 import { exec } from './common';
-import type { RawExecOptions } from './types';
+import type { DataListener, RawExecOptions } from './types';
 
 jest.mock('node:child_process');
 const spawn = mockedFunction(_spawn);
@@ -147,6 +147,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 +178,48 @@ 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',
+            outputListeners: {
+              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,8 +1,10 @@
 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';
+import type { DataListener, ExecResult, RawExecOptions } from './types';
 
 // https://man7.org/linux/man-pages/man7/signal.7.html#NAME
 // Non TERM/CORE signals
@@ -34,6 +36,9 @@ function initStreamListeners(
   let stdoutLen = 0;
   let stderrLen = 0;
 
+  registerDataListeners(cp.stdout, opts.outputListeners?.stdout);
+  registerDataListeners(cp.stderr, opts.outputListeners?.stderr);
+
   cp.stdout?.on('data', (chunk: Buffer) => {
     // process.stdout.write(data.toString());
     const len = Buffer.byteLength(chunk, encoding);
@@ -58,6 +63,19 @@ function initStreamListeners(
   return [stdout, stderr];
 }
 
+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);
+  }
+}
+
 export function exec(cmd: string, opts: RawExecOptions): Promise<ExecResult> {
   return new Promise((resolve, reject) => {
     const maxBuffer = opts.maxBuffer ?? 10 * 1024 * 1024; // Set default max buffer size to 10MB

lib/util/exec/types.ts

@@ -25,6 +25,12 @@ export interface DockerOptions {
   cwd?: Opt<string>;
 }
 
+export type DataListener = (chunk: any) => void;
+export type OutputListeners = {
+  stdout?: DataListener[];
+  stderr?: DataListener[];
+};
+
 export interface RawExecOptions extends ChildProcessSpawnOptions {
   // TODO: to be removed in #16655
   /**
@@ -33,6 +39,7 @@ export interface RawExecOptions extends ChildProcessSpawnOptions {
   encoding: string;
   maxBuffer?: number | undefined;
   cwd?: string;
+  outputListeners?: OutputListeners;
 }
 
 export interface ExecResult {