0
# Stream Management and Debugging
1

2
Advanced stream handling, output transformation, and interactive debugging capabilities for WebdriverIO Local Runner.
3

4
## Capabilities
5

6
### Stream Transformation
7

8
Transform and aggregate worker process output streams with capability ID prefixing and debugging message filtering.
9

10
```typescript { .api }
11
/**
12
 * Transform worker stream output with CID prefixing and debug filtering
13
 * @param cid - Capability ID for prefixing output
14
 * @param inputStream - Input stream from worker process
15
 * @param aggregator - Optional array to collect output for later use
16
 * @returns Transformed readable stream
17
 */
18
function runnerTransformStream(
19
  cid: string,
20
  inputStream: Readable,
21
  aggregator?: string[]
22
): Readable;
23
```
24

25
**Usage Example:**
26

27
```typescript
28
import { runnerTransformStream } from "@wdio/local-runner";
29
import { Readable } from "stream";
30

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

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

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

42
// Access aggregated logs
43
console.log("Collected logs:", logAggregator);
44
```
45

46
### Runner Stream Class
47

48
Custom Transform stream implementation for handling runner output with proper buffering and event management.
49

50
```typescript { .api }
51
/**
52
 * Custom Transform stream for runner output handling
53
 * Manages proper event propagation and buffering
54
 */
55
class RunnerStream extends Transform {
56
  // Transform stream implementation for runner output
57
}
58
```
59

60
**Usage Example:**
61

62
```typescript
63
import { RunnerStream } from "@wdio/local-runner";
64

65
const runnerOutput = new RunnerStream();
66

67
// Pipe worker streams through runner stream
68
workerTransformStream.pipe(runnerOutput);
69
runnerOutput.pipe(process.stdout);
70
```
71

72
### REPL Debugging System
73

74
Interactive debugging capabilities with queue management for handling multiple worker debug sessions.
75

76
```typescript { .api }
77
/**
78
 * REPL implementation for interactive debugging
79
 */
80
class WDIORunnerRepl {
81
  onResult(params: any): void;
82
  // Additional REPL methods for interactive debugging
83
}
84

85
/**
86
 * Queue manager for handling multiple REPL sessions
87
 */
88
class ReplQueue {
89
  isRunning: boolean;
90
  runningRepl?: WDIORunnerRepl;
91
  
92
  /**
93
   * Add REPL session to queue
94
   * @param childProcess - Worker child process for IPC
95
   * @param config - REPL configuration with prompt and options
96
   * @param onStart - Callback when REPL session starts
97
   * @param onResult - Callback for REPL results
98
   */
99
  add(
100
    childProcess: ChildProcess,
101
    config: ReplConfig,
102
    onStart: () => void,
103
    onResult: (ev: unknown) => void
104
  ): void;
105
  
106
  /**
107
   * Process next REPL session in queue
108
   */
109
  next(): void;
110
}
111
```
112

113
**Usage Example:**
114

115
```typescript
116
import { ReplQueue, WDIORunnerRepl } from "@wdio/local-runner";
117

118
const replQueue = new ReplQueue();
119

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

128
// Process debug sessions
129
replQueue.next();
130

131
// Check if REPL is currently running
132
if (replQueue.isRunning) {
133
  console.log("Debug session in progress");
134
}
135
```
136

137
### Constants and Configuration
138

139
Core constants used throughout the local runner for timeouts, buffering, and debugging.
140

141
```typescript { .api }
142
/**
143
 * Timeout for graceful worker shutdown (milliseconds)
144
 */
145
const SHUTDOWN_TIMEOUT: 5000;
146

147
/**
148
 * Array of debugger message patterns to filter from output
149
 */
150
const DEBUGGER_MESSAGES: string[];
151

152
/**
153
 * Stream buffer configuration options
154
 */
155
const BUFFER_OPTIONS: {
156
  initialSize: number;      // 1000 * 1024 (1 MB)
157
  incrementAmount: number;  // 100 * 1024 (100 KB)
158
};
159
```
160

161
**Usage Example:**
162

163
```typescript
164
import { SHUTDOWN_TIMEOUT, BUFFER_OPTIONS } from "@wdio/local-runner";
165
import { WritableStreamBuffer } from "stream-buffers";
166

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

171
// Use shutdown timeout for graceful termination
172
const shutdownPromise = new Promise((resolve) => {
173
  setTimeout(resolve, SHUTDOWN_TIMEOUT);
174
});
175
```
176

177
### Utility Functions
178

179
Helper functions for stream and event management.
180

181
```typescript { .api }
182
/**
183
 * Remove the last listener of a specific event from a Transform stream
184
 * @param target - Transform stream target
185
 * @param eventName - Name of event to remove listener from
186
 */
187
function removeLastListener(target: Transform, eventName: string): void;
188
```
189

190
**Usage Example:**
191

192
```typescript
193
import { removeLastListener } from "@wdio/local-runner";
194
import { Transform } from "stream";
195

196
const transformStream = new Transform();
197

198
// Add multiple listeners
199
transformStream.on('data', handler1);
200
transformStream.on('data', handler2);
201
transformStream.on('data', handler3);
202

203
// Remove the last 'data' listener (handler3)
204
removeLastListener(transformStream, 'data');
205
```
206

207
## Advanced Features
208

209
### Debugger Message Filtering
210

211
Worker output streams automatically filter debugger-related messages to keep test output clean.
212

213
```typescript
214
// These messages are automatically filtered from worker output:
215
// - "Debugger listening on"
216
// - "Debugger attached" 
217
// - "Waiting for the debugger"
218

219
// Filtering is handled automatically by runnerTransformStream
220
const cleanOutput = runnerTransformStream(cid, workerStream);
221
```
222

223
### Output Aggregation
224

225
Collect worker output for later analysis or reporting.
226

227
```typescript
228
// Aggregate output by test spec for grouped logging
229
const specLogs: string[] = [];
230
const groupedStream = runnerTransformStream(cid, workerStream, specLogs);
231

232
// Output is collected in specLogs array for later use
233
// This enables groupLogsByTestSpec functionality
234
```
235

236
### Multiple REPL Session Management
237

238
Handle concurrent debug sessions from multiple workers.
239

240
```typescript
241
const replQueue = new ReplQueue();
242

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

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

256
// Queue ensures only one debug session runs at a time
257
replQueue.next();
258
```
259

260
## Error Handling
261

262
### Stream Error Management
263

264
```typescript
265
transformedStream.on('error', (error) => {
266
  console.error('Stream transformation error:', error);
267
  // Handle stream errors gracefully
268
});
269
```
270

271
### REPL Session Errors
272

273
```typescript
274
replQueue.add(
275
  childProcess,
276
  config,
277
  () => console.log('REPL started'),
278
  (result) => {
279
    if (result instanceof Error) {
280
      console.error('REPL error:', result.message);
281
    } else {
282
      console.log('REPL result:', result);
283
    }
284
  }
285
);
286
```
287

288
### Buffer Overflow Protection
289

290
```typescript
291
// Buffer options provide overflow protection
292
const bufferConfig = {
293
  initialSize: BUFFER_OPTIONS.initialSize,     // Start with 1MB
294
  incrementAmount: BUFFER_OPTIONS.incrementAmount, // Grow by 100KB chunks
295
  maxSize: 10 * 1024 * 1024 // Optional: Set maximum size (10MB)
296
};
297

298
const protectedBuffer = new WritableStreamBuffer(bufferConfig);
299
```