Support stdio

fixtures/asynchronous-modules/stdio.js

@@ -0,0 +1,14 @@
+export const console = {...globalThis.console}
+export const logs = async (messages) => {
+  for (const {type, message} of messages) {
+    console[type](message)
+  }
+}
+export const writes = async (messages) => {
+  for (const {type, message} of messages) {
+    process[type].write(message)
+  }
+}
+export const printObject = async () => {
+  console.log(printObject)
+}

package.json

@@ -79,6 +79,7 @@
     "npm-run-all": "4.1.5",
     "prettier": "3.2.4",
     "sort-package-json": "2.6.0",
+    "strip-ansi": "7.1.0",
     "tempy": "3.1.0"
   },
   "packageManager": "[email protected]",

source/response.js

@@ -10,27 +10,40 @@ class Response {
 
   #actionHandlers
 
+  #stdio = []
+
   constructor(actionHandlers) {
     this.#actionHandlers = actionHandlers
 
     process.exit = () => {
       this.#terminate()
       processExit()
     }
+
+    // https://github.com/nodejs/node/blob/66556f53a7b36384bce305865c30ca43eaa0874b/lib/internal/worker/io.js#L369
+    for (const type of ['stdout', 'stderr']) {
+      process[type]._writev = (chunks, callback) => {
+        for (const {chunk} of chunks) {
+          this.#stdio.push({type, chunk})
+        }
+
+        callback()
+      }
+    }
   }
 
   #send(response) {
     const responsePort = this.#responsePort
     const signal = this.#signal
 
     try {
-      responsePort.postMessage(response)
+      responsePort.postMessage({...response, stdio: this.#stdio})
     } catch {
       const error = new Error(
         `Cannot serialize worker response:\n${util.inspect(response.result)}`,
       )
 
-      responsePort.postMessage({error})
+      responsePort.postMessage({error, stdio: this.#stdio})
     } finally {
       responsePort.close()
 
@@ -68,6 +81,7 @@ class Response {
       async ({signal, port, action, payload}) => {
         this.#signal = signal
         this.#responsePort = port
+        this.#stdio.length = 0
 
         try {
           this.#sendResult(await this.#processAction(action, payload))

source/threads-worker.js

@@ -64,13 +64,17 @@ class ThreadsWorker {
   @param {number} [timeout]
   */
   #sendActionToWorker(worker, action, payload, timeout) {
-    const {terminated, result, error, errorData} = request(
+    const {terminated, result, error, errorData, stdio} = request(
       worker,
       action,
       payload,
       timeout,
     )
 
+    for (const {chunk, type} of stdio) {
+      process[type].write(chunk)
+    }
+
     if (terminated && this.#worker) {
       this.#worker.terminate()
       this.#worker = undefined

tests/stdio.test.js

@@ -0,0 +1,136 @@
+import test from 'node:test'
+import * as assert from 'node:assert/strict'
+import process from 'node:process'
+import stripAnsi from 'strip-ansi'
+import loadModuleForTests from '../scripts/load-module-for-tests.js'
+
+const {makeSynchronized} = await loadModuleForTests()
+
+const stdio = makeSynchronized(
+  new URL('../fixtures/asynchronous-modules/stdio.js', import.meta.url),
+)
+
+const getResult = (run) => {
+  const original = {}
+  const result = []
+
+  for (const type of ['stdout', 'stderr']) {
+    original[type] = process[type].write
+
+    process[type].write = (message) => {
+      message = stripAnsi(message)
+      result.push({type, message})
+    }
+  }
+
+  try {
+    run()
+  } finally {
+    for (const type of ['stdout', 'stderr']) {
+      process[type].write = original[type]
+    }
+  }
+
+  return result
+}
+
+test('stdio', () => {
+  {
+    const result = getResult(() => {
+      stdio.console.log('console.log called')
+    })
+    assert.deepEqual(result, [
+      {type: 'stdout', message: 'console.log called\n'},
+    ])
+  }
+
+  {
+    const result = getResult(() => {
+      stdio.console.warn('console.warn called')
+    })
+    assert.deepEqual(result, [
+      {type: 'stderr', message: 'console.warn called\n'},
+    ])
+  }
+
+  {
+    const result = getResult(() => {
+      stdio.console.error('console.error called')
+    })
+    assert.deepEqual(result, [
+      {type: 'stderr', message: 'console.error called\n'},
+    ])
+  }
+
+  {
+    const result = getResult(() => {
+      stdio.console.table([{fisker: 'jerk'}])
+    })
+    const table = /* Indent */ `
+      ┌─────────┬────────┐
+      │ (index) │ fisker │
+      ├─────────┼────────┤
+      │    0    │ 'jerk' │
+      └─────────┴────────┘
+    `
+    assert.deepEqual(result, [
+      {
+        type: 'stdout',
+        message: `${table
+          .trim()
+          .split('\n')
+          .map((line) => line.trim())
+          .join('\n')}\n`,
+      },
+    ])
+  }
+
+  {
+    const result = getResult(() => {
+      stdio.logs([
+        {type: 'log', message: '1'},
+        {type: 'error', message: '2'},
+        {type: 'log', message: '3'},
+        {type: 'warn', message: '4'},
+      ])
+    })
+    assert.deepEqual(result, [
+      {type: 'stdout', message: '1\n'},
+      {type: 'stderr', message: '2\n'},
+      {type: 'stdout', message: '3\n'},
+      {type: 'stderr', message: '4\n'},
+    ])
+  }
+
+  {
+    const [{message}] = getResult(() => {
+      stdio.logs([{type: 'time'}, {type: 'timeEnd'}])
+    })
+
+    assert.ok(/default: [\d.]+ms\n/.test(message), message)
+  }
+
+  {
+    const messages = [
+      {type: 'stdout', message: '1'},
+      {type: 'stderr', message: '2'},
+      {type: 'stdout', message: '3'},
+      {type: 'stderr', message: '4'},
+    ]
+
+    const result = getResult(() => {
+      stdio.writes(messages)
+    })
+    assert.deepEqual(result, messages)
+  }
+
+  {
+    const result = getResult(() => {
+      stdio.printObject()
+    })
+
+    assert.deepEqual(result, [
+      {type: 'stdout', message: '[AsyncFunction: printObject]\n'},
+    ])
+  }
+})