Stream Management and Debugging

/**
 * Transform worker stream output with CID prefixing and debug filtering
 * @param cid - Capability ID for prefixing output
 * @param inputStream - Input stream from worker process
 * @param aggregator - Optional array to collect output for later use
 * @returns Transformed readable stream
 */
function runnerTransformStream(
  cid: string,
  inputStream: Readable,
  aggregator?: string[]
): Readable;

import { runnerTransformStream } from "@wdio/local-runner";
import { Readable } from "stream";

// Transform worker stdout with capability ID prefixing
const workerStdout: Readable = childProcess.stdout;
const transformedStream = runnerTransformStream("0-0", workerStdout);

// With output aggregation for later retrieval
const logAggregator: string[] = [];
const aggregatedStream = runnerTransformStream("0-0", workerStdout, logAggregator);

// Pipe to main process stdout
transformedStream.pipe(process.stdout);

// Access aggregated logs
console.log("Collected logs:", logAggregator);

/**
 * Custom Transform stream for runner output handling
 * Manages proper event propagation and buffering
 */
class RunnerStream extends Transform {
  // Transform stream implementation for runner output
}

import { RunnerStream } from "@wdio/local-runner";

const runnerOutput = new RunnerStream();

// Pipe worker streams through runner stream
workerTransformStream.pipe(runnerOutput);
runnerOutput.pipe(process.stdout);

/**
 * REPL implementation for interactive debugging
 */
class WDIORunnerRepl {
  onResult(params: any): void;
  // Additional REPL methods for interactive debugging
}

/**
 * Queue manager for handling multiple REPL sessions
 */
class ReplQueue {
  isRunning: boolean;
  runningRepl?: WDIORunnerRepl;
  
  /**
   * Add REPL session to queue
   * @param childProcess - Worker child process for IPC
   * @param config - REPL configuration with prompt and options
   * @param onStart - Callback when REPL session starts
   * @param onResult - Callback for REPL results
   */
  add(
    childProcess: ChildProcess,
    config: ReplConfig,
    onStart: () => void,
    onResult: (ev: unknown) => void
  ): void;
  
  /**
   * Process next REPL session in queue
   */
  next(): void;
}

import { ReplQueue, WDIORunnerRepl } from "@wdio/local-runner";

const replQueue = new ReplQueue();

// Add debug session to queue
replQueue.add(
  workerProcess,
  { prompt: "[0-0] › ", timeout: 30000 },
  () => console.log("Debug session started"),
  (result) => console.log("Debug result:", result)
);

// Process debug sessions
replQueue.next();

// Check if REPL is currently running
if (replQueue.isRunning) {
  console.log("Debug session in progress");
}

/**
 * Timeout for graceful worker shutdown (milliseconds)
 */
const SHUTDOWN_TIMEOUT: 5000;

/**
 * Array of debugger message patterns to filter from output
 */
const DEBUGGER_MESSAGES: string[];

/**
 * Stream buffer configuration options
 */
const BUFFER_OPTIONS: {
  initialSize: number;      // 1000 * 1024 (1 MB)
  incrementAmount: number;  // 100 * 1024 (100 KB)
};

import { SHUTDOWN_TIMEOUT, BUFFER_OPTIONS } from "@wdio/local-runner";
import { WritableStreamBuffer } from "stream-buffers";

// Use buffer options for creating output streams
const stdout = new WritableStreamBuffer(BUFFER_OPTIONS);
const stderr = new WritableStreamBuffer(BUFFER_OPTIONS);

// Use shutdown timeout for graceful termination
const shutdownPromise = new Promise((resolve) => {
  setTimeout(resolve, SHUTDOWN_TIMEOUT);
});

/**
 * Remove the last listener of a specific event from a Transform stream
 * @param target - Transform stream target
 * @param eventName - Name of event to remove listener from
 */
function removeLastListener(target: Transform, eventName: string): void;

import { removeLastListener } from "@wdio/local-runner";
import { Transform } from "stream";

const transformStream = new Transform();

// Add multiple listeners
transformStream.on('data', handler1);
transformStream.on('data', handler2);
transformStream.on('data', handler3);

// Remove the last 'data' listener (handler3)
removeLastListener(transformStream, 'data');

// These messages are automatically filtered from worker output:
// - "Debugger listening on"
// - "Debugger attached" 
// - "Waiting for the debugger"

// Filtering is handled automatically by runnerTransformStream
const cleanOutput = runnerTransformStream(cid, workerStream);

// Aggregate output by test spec for grouped logging
const specLogs: string[] = [];
const groupedStream = runnerTransformStream(cid, workerStream, specLogs);

// Output is collected in specLogs array for later use
// This enables groupLogsByTestSpec functionality

const replQueue = new ReplQueue();

// Multiple workers can request debug sessions
worker1.on('message', (payload) => {
  if (payload.name === 'start' && payload.origin === 'debugger') {
    replQueue.add(worker1.childProcess, payload.params, startHandler, resultHandler);
  }
});

worker2.on('message', (payload) => {
  if (payload.name === 'start' && payload.origin === 'debugger') {
    replQueue.add(worker2.childProcess, payload.params, startHandler, resultHandler);
  }
});

// Queue ensures only one debug session runs at a time
replQueue.next();

transformedStream.on('error', (error) => {
  console.error('Stream transformation error:', error);
  // Handle stream errors gracefully
});

replQueue.add(
  childProcess,
  config,
  () => console.log('REPL started'),
  (result) => {
    if (result instanceof Error) {
      console.error('REPL error:', result.message);
    } else {
      console.log('REPL result:', result);
    }
  }
);

// Buffer options provide overflow protection
const bufferConfig = {
  initialSize: BUFFER_OPTIONS.initialSize,     // Start with 1MB
  incrementAmount: BUFFER_OPTIONS.incrementAmount, // Grow by 100KB chunks
  maxSize: 10 * 1024 * 1024 // Optional: Set maximum size (10MB)
};

const protectedBuffer = new WritableStreamBuffer(bufferConfig);