NodeJS Compat: Add process.stdin, stdout and stderr (#4480)

Author: guybedford

src/node/internal/internal_fs_streams.ts

@@ -368,7 +368,7 @@ export declare class ReadStream extends Readable {
   [kFs]: RealizedFsOperations;
   [kIsPerformingIO]: boolean;
   [kHandle]: FileHandle | undefined;
-  constructor(path: FilePath, options?: ReadStreamOptions);
+  constructor(path: FilePath | null, options?: ReadStreamOptions);
   push(chunk: NodeJS.ArrayBufferView | null): boolean;
   close(callback?: ErrorOnlyCallback): void;
 }

src/node/internal/public_process.ts

@@ -16,12 +16,11 @@ import {
   ERR_UNSUPPORTED_OPERATION,
 } from 'node-internal:internal_errors';
 import processImpl from 'node-internal:process';
-import type { Buffer } from 'node:buffer';
+import { Buffer } from 'node-internal:internal_buffer';
 import { parseEnv } from 'node-internal:internal_utils';
-import type * as NodeFS from 'node:fs';
-
-const { compatibilityFlags } = Cloudflare;
-
+import { Writable } from 'node-internal:streams_writable';
+import { writeSync } from 'node-internal:internal_fs_sync';
+import { ReadStream } from 'node-internal:internal_fs_streams';
 import {
   platform,
   nextTick,
@@ -32,12 +31,63 @@ import {
 } from 'node-internal:internal_process';
 import { validateString } from 'node-internal:validators';
 
+import type { Readable } from 'node-internal:streams_readable';
+import type * as NodeFS from 'node:fs';
+
+const { compatibilityFlags } = Cloudflare;
+
 export { platform, nextTick, emitWarning, env, features };
 
-// TODO(soon): Implement stdio along with TTY streams (and as a requirement for removing experimental).
-export const stdin = undefined;
-export const stdout = undefined;
-export const stderr = undefined;
+// For stdin, we emulate `node foo.js < /dev/null`
+export const stdin = new ReadStream(null, {
+  fd: 0,
+  autoClose: false,
+}) as Readable & {
+  fd: number;
+};
+stdin.fd = 0;
+
+function chunkToBuffer(
+  chunk: Buffer | ArrayBufferView | DataView | string,
+  encoding: BufferEncoding
+): Buffer {
+  if (Buffer.isBuffer(chunk)) {
+    return chunk;
+  }
+  if (typeof chunk === 'string') {
+    return Buffer.from(chunk, encoding);
+  }
+  return Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength);
+}
+
+// For stdout, we emulate `nohup node foo.js`
+class SyncWriteStream extends Writable {
+  fd: number;
+  readable: boolean;
+  _type = 'fs';
+  _isStdio = true;
+  constructor(fd: number) {
+    super({ autoDestroy: true });
+    this.fd = fd;
+    this.readable = false;
+  }
+  override _write(
+    chunk: string | Buffer | ArrayBufferView | DataView,
+    encoding: BufferEncoding,
+    cb: (error?: Error | null) => void
+  ): void {
+    try {
+      writeSync(this.fd, chunkToBuffer(chunk, encoding));
+    } catch (e: unknown) {
+      cb(e as Error);
+      return;
+    }
+    cb();
+  }
+}
+
+export const stdout = new SyncWriteStream(1);
+export const stderr = new SyncWriteStream(2);
 
 export function chdir(path: string | Buffer | URL): void {
   validateString(path, 'directory');
@@ -200,8 +250,6 @@ export function uptime(): number {
   return 0;
 }
 
-// TODO(soon): Support with proper process.cwd() resolution along with
-//             test in process-nodejs-test.
 export function loadEnvFile(path: string | URL | Buffer = '.env'): void {
   if (!compatibilityFlags.experimental || !compatibilityFlags.nodejs_compat) {
     throw new ERR_UNSUPPORTED_OPERATION();

src/node/internal/streams_adapters.js

@@ -41,6 +41,7 @@ import {
 } from 'node-internal:streams_util';
 
 import { Buffer } from 'node-internal:internal_buffer';
+import { nextTick } from 'node-internal:internal_process';
 
 import {
   ERR_INVALID_ARG_TYPE,
@@ -53,8 +54,6 @@ import { normalizeEncoding } from 'node-internal:internal_utils';
 
 import { validateBoolean, validateObject } from 'node-internal:validators';
 
-import process from 'node:process';
-
 const encoder = new TextEncoder();
 
 /**
@@ -220,7 +219,7 @@ export function newStreamWritableFromWritableStream(
           // thrown we don't want those to cause an unhandled
           // rejection. Let's just escape the promise and
           // handle it separately.
-          process.nextTick(() => destroy.call(writable, error));
+          nextTick(() => destroy.call(writable, error));
         }
       }
 
@@ -271,7 +270,7 @@ export function newStreamWritableFromWritableStream(
           // thrown we don't want those to cause an unhandled
           // rejection. Let's just escape the promise and
           // handle it separately.
-          process.nextTick(() => {
+          nextTick(() => {
             throw error;
           });
         }
@@ -299,7 +298,7 @@ export function newStreamWritableFromWritableStream(
           // thrown we don't want those to cause an unhandled
           // rejection. Let's just escape the promise and
           // handle it separately.
-          process.nextTick(() => destroy.call(writable, error));
+          nextTick(() => destroy.call(writable, error));
         }
       }
 
@@ -491,7 +490,7 @@ export function newStreamReadableFromReadableStream(
           // thrown we don't want those to cause an unhandled
           // rejection. Let's just escape the promise and
           // handle it separately.
-          process.nextTick(() => {
+          nextTick(() => {
             throw error;
           });
         }
@@ -641,7 +640,7 @@ export function newStreamDuplexFromReadableWritablePair(
           // thrown we don't want those to cause an unhandled
           // rejection. Let's just escape the promise and
           // handle it separately.
-          process.nextTick(() => destroy.call(duplex, error));
+          nextTick(() => destroy.call(duplex, error));
         }
       }
 
@@ -693,7 +692,7 @@ export function newStreamDuplexFromReadableWritablePair(
           // thrown we don't want those to cause an unhandled
           // rejection. Let's just escape the promise and
           // handle it separately.
-          process.nextTick(() => destroy.call(duplex, error));
+          nextTick(() => destroy.call(duplex, error));
         }
       }
 
@@ -725,7 +724,7 @@ export function newStreamDuplexFromReadableWritablePair(
           // thrown we don't want those to cause an unhandled
           // rejection. Let's just escape the promise and
           // handle it separately.
-          process.nextTick(() => {
+          nextTick(() => {
             throw error;
           });
         }

src/node/internal/streams_duplex.js

@@ -34,8 +34,7 @@ import {
   newStreamDuplexFromReadableWritablePair,
   newReadableWritablePairFromDuplex,
 } from 'node-internal:streams_adapters';
-
-import process from 'node:process';
+import { nextTick } from 'node-internal:internal_process';
 
 import {
   addAbortSignal,
@@ -322,9 +321,9 @@ function duplexify(body, name) {
           final(async () => {
             try {
               await promise;
-              process.nextTick(cb, null);
+              nextTick(cb, null);
             } catch (err) {
-              process.nextTick(cb, err);
+              nextTick(cb, err);
             }
           });
         },
@@ -434,7 +433,7 @@ function fromAsyncGen(fn) {
         const _promise = promise;
         promise = null;
         const { chunk, done, cb } = await _promise;
-        process.nextTick(cb);
+        nextTick(cb);
         if (done) return;
         if (signal.aborted)
           throw new AbortError(undefined, {
@@ -633,7 +632,7 @@ class DuplexSide extends Duplex {
     assert(this.#otherSide !== null);
     assert(this.#otherSide[kCallback] === null);
     if (chunk.length === 0) {
-      process.nextTick(callback);
+      nextTick(callback);
     } else {
       this.#otherSide.push(chunk);
       this.#otherSide[kCallback] = callback;

src/node/internal/streams_pipeline.js

@@ -36,7 +36,7 @@ import {
   destroyer as destroyerImpl,
 } from 'node-internal:streams_util';
 
-import process from 'node:process';
+import { nextTick } from 'node-internal:internal_process';
 
 import { PassThrough } from 'node-internal:streams_transform';
 import { Duplex } from 'node-internal:streams_duplex';
@@ -214,7 +214,7 @@ export function pipelineImpl(streams, callback, opts) {
       if (!error) {
         lastStreamCleanup.forEach((fn) => fn());
       }
-      process.nextTick(callback, error, value);
+      nextTick(callback, error, value);
     }
   }
   let ret;
@@ -309,11 +309,11 @@ export function pipelineImpl(streams, callback, opts) {
               if (end) {
                 pt.end();
               }
-              process.nextTick(finish);
+              nextTick(finish);
             },
             (err) => {
               pt.destroy(err);
-              process.nextTick(finish, err);
+              nextTick(finish, err);
             }
           );
         } else if (isIterable(ret, true)) {
@@ -365,7 +365,7 @@ export function pipelineImpl(streams, callback, opts) {
     (signal !== null && signal !== undefined && signal.aborted) ||
     (outerSignal !== null && outerSignal !== undefined && outerSignal.aborted)
   ) {
-    process.nextTick(abort);
+    nextTick(abort);
   }
   return ret;
 }

src/node/internal/streams_readable.js

@@ -42,8 +42,7 @@ import {
   finished,
   kOnConstructed,
 } from 'node-internal:streams_util';
-
-import process from 'node:process';
+import { nextTick } from 'node-internal:internal_process';
 
 import { EventEmitter } from 'node-internal:events';
 
@@ -547,7 +546,7 @@ function emitReadable(stream) {
   state.needReadable = false;
   if (!state.emittedReadable) {
     state.emittedReadable = true;
-    process.nextTick(emitReadable_, stream);
+    nextTick(emitReadable_, stream);
   }
 }
 
@@ -578,7 +577,7 @@ function emitReadable_(stream) {
 function maybeReadMore(stream, state) {
   if (!state.readingMore && state.constructed) {
     state.readingMore = true;
-    process.nextTick(maybeReadMore_, stream, state);
+    nextTick(maybeReadMore_, stream, state);
   }
 }
 
@@ -643,7 +642,7 @@ Readable.prototype.pipe = function (dest, pipeOpts) {
   state.pipes.push(dest);
   const doEnd = !pipeOpts || pipeOpts.end !== false;
   const endFn = doEnd ? onend : unpipe;
-  if (state.endEmitted) process.nextTick(endFn);
+  if (state.endEmitted) nextTick(endFn);
   else src.once('end', endFn);
   dest.on('unpipe', onunpipe);
   function onunpipe(readable, unpipeInfo) {
@@ -835,7 +834,7 @@ Readable.prototype.on = function (ev, fn) {
       if (state.length) {
         emitReadable(this);
       } else if (!state.reading) {
-        process.nextTick(nReadingNextTick, this);
+        nextTick(nReadingNextTick, this);
       }
     }
   }
@@ -851,7 +850,7 @@ Readable.prototype.removeListener = function (ev, fn) {
     // support once('readable', fn) cycles. This means that calling
     // resume within the same tick will have no
     // effect.
-    process.nextTick(updateReadableListening, this);
+    nextTick(updateReadableListening, this);
   }
   return res;
 };
@@ -865,7 +864,7 @@ Readable.prototype.removeAllListeners = function (ev) {
     // support once('readable', fn) cycles. This means that calling
     // resume within the same tick will have no
     // effect.
-    process.nextTick(updateReadableListening, this);
+    nextTick(updateReadableListening, this);
   }
   return res;
 };
@@ -908,7 +907,7 @@ Readable.prototype.resume = function () {
 function resume(stream, state) {
   if (!state.resumeScheduled) {
     state.resumeScheduled = true;
-    process.nextTick(resume_, stream, state);
+    nextTick(resume_, stream, state);
   }
 }
 
@@ -1225,7 +1224,7 @@ function endReadable(stream) {
   const state = stream._readableState;
   if (!state.endEmitted) {
     state.ended = true;
-    process.nextTick(endReadableNT, state, stream);
+    nextTick(endReadableNT, state, stream);
   }
 }
 
@@ -1240,7 +1239,7 @@ function endReadableNT(state, stream) {
     state.endEmitted = true;
     stream.emit('end');
     if (stream.writable && stream.allowHalfOpen === false) {
-      process.nextTick(endWritableNT, stream);
+      nextTick(endWritableNT, stream);
     } else if (state.autoDestroy) {
       // In case of duplex streams we need a way to detect
       // if the writable side is ready for autoDestroy as well.
@@ -1344,8 +1343,8 @@ export function from(Readable, iterable, opts) {
   };
   readable._destroy = function (error, cb) {
     close(error).then(
-      () => process.nextTick(cb, error),
-      (err) => process.nextTick(cb, err || error)
+      () => nextTick(cb, error),
+      (err) => nextTick(cb, err || error)
     );
   };
   async function close(error) {