feat: allow providing a PathRef for .stdin(...) (#211)

mod.test.ts

@@ -754,11 +754,11 @@ Deno.test("piping to stdin", async () => {
 
   // string
   {
-    const result =
-      await $`deno eval "const b = new Uint8Array(4); await Deno.stdin.read(b); await Deno.stdout.write(b);"`
-        .stdinText("test\n")
-        .text();
-    assertEquals(result, "test");
+    const command = $`deno eval "const b = new Uint8Array(4); await Deno.stdin.read(b); await Deno.stdout.write(b);"`
+      .stdinText("test\n");
+    // should support calling multiple times
+    assertEquals(await command.text(), "test");
+    assertEquals(await command.text(), "test");
   }
 
   // Uint8Array
@@ -778,6 +778,15 @@ Deno.test("piping to stdin", async () => {
       .text();
     assertEquals(result, "1\n2");
   }
+
+  // PathRef
+  await withTempDir(async (tempDir) => {
+    const tempFile = tempDir.join("temp_file.txt");
+    const fileText = "1 testing this out\n".repeat(1_000);
+    tempFile.writeTextSync(fileText);
+    const output = await $`cat`.stdin(tempFile).text();
+    assertEquals(output, fileText.trim());
+  });
 });
 
 Deno.test("spawning a command twice that has stdin set to a Reader should error", async () => {

src/command.ts

@@ -33,9 +33,24 @@ import { PathRef } from "./path.ts";
 
 type BufferStdio = "inherit" | "null" | "streamed" | Buffer;
 
+class Deferred<T> {
+  #create: () => T | Promise<T>;
+  constructor(create: () => T | Promise<T>) {
+    this.#create = create;
+  }
+
+  create() {
+    return this.#create();
+  }
+}
+
 interface CommandBuilderState {
   command: string | undefined;
-  stdin: "inherit" | "null" | Box<Reader | ReadableStream<Uint8Array> | "consumed">;
+  stdin:
+    | "inherit"
+    | "null"
+    | Box<Reader | ReadableStream<Uint8Array> | "consumed">
+    | Deferred<ReadableStream<Uint8Array> | Reader>;
   combinedStdoutStderr: boolean;
   stdoutKind: ShellPipeWriterKind;
   stderrKind: ShellPipeWriterKind;
@@ -254,12 +269,17 @@ export class CommandBuilder implements PromiseLike<CommandResult> {
    * For this reason, if you are setting stdin to something other than "inherit" or
    * "null", then it's recommended to set this each time you spawn a command.
    */
-  stdin(reader: ShellPipeReader | Uint8Array | ReadableStream<Uint8Array>): CommandBuilder {
+  stdin(reader: ShellPipeReader): CommandBuilder {
     return this.#newWithState((state) => {
       if (reader === "inherit" || reader === "null") {
         state.stdin = reader;
       } else if (reader instanceof Uint8Array) {
-        state.stdin = new Box(new Buffer(reader));
+        state.stdin = new Deferred(() => new Buffer(reader));
+      } else if (reader instanceof PathRef) {
+        state.stdin = new Deferred(async () => {
+          const file = await reader.open();
+          return file.readable;
+        });
       } else {
         state.stdin = new Box(reader);
       }
@@ -612,7 +632,7 @@ export function parseAndSpawnCommand(state: CommandBuilderState) {
   return new CommandChild(async (resolve, reject) => {
     try {
       const list = parseCommand(command);
-      const stdin = takeStdin();
+      const stdin = await takeStdin();
       let code = await spawn(list, {
         stdin: stdin instanceof ReadableStream ? readerFromStreamReader(stdin.getReader()) : stdin,
         stdout,
@@ -668,7 +688,7 @@ export function parseAndSpawnCommand(state: CommandBuilderState) {
     killSignalController,
   });
 
-  function takeStdin() {
+  async function takeStdin() {
     if (state.stdin instanceof Box) {
       const stdin = state.stdin.value;
       if (stdin === "consumed") {
@@ -680,6 +700,8 @@ export function parseAndSpawnCommand(state: CommandBuilderState) {
       }
       state.stdin.value = "consumed";
       return stdin;
+    } else if (state.stdin instanceof Deferred) {
+      return await state.stdin.create();
     } else {
       return state.stdin;
     }

src/deps.ts

@@ -3,7 +3,7 @@ export * as fs from "https://deno.land/[email protected]/fs/mod.ts";
 export { Buffer } from "https://deno.land/[email protected]/io/buffer.ts";
 export { BufReader } from "https://deno.land/[email protected]/io/buf_reader.ts";
 export * as path from "https://deno.land/[email protected]/path/mod.ts";
-export { readAll } from "https://deno.land/[email protected]/streams/read_all.ts";
+export { readAll } from "https://deno.land/[email protected]/io/read_all.ts";
 export { readerFromStreamReader } from "https://deno.land/[email protected]/streams/reader_from_stream_reader.ts";
 export { writeAll, writeAllSync } from "https://deno.land/[email protected]/io/write_all.ts";
 export { outdent } from "https://deno.land/x/[email protected]/src/index.ts";

src/pipes.ts

@@ -1,3 +1,4 @@
+import { PathRef } from "./path.ts";
 import { logger } from "./console/logger.ts";
 import { Buffer, writeAllSync } from "./deps.ts";
 
@@ -15,7 +16,13 @@ export interface Closer {
   close(): void;
 }
 
-export type ShellPipeReader = "inherit" | "null" | Reader;
+export type ShellPipeReader =
+  | "inherit"
+  | "null"
+  | Reader
+  | ReadableStream<Uint8Array>
+  | Uint8Array
+  | PathRef;
 /**
  * The behaviour to use for a shell pipe.
  * @value "inherit" - Sends the output directly to the current process' corresponding pipe (default).

src/request.test.ts

@@ -10,11 +10,12 @@ function withServer(action: (serverUrl: URL) => Promise<void>) {
         const url = new URL(`http://${details.hostname}:${details.port}/`);
         try {
           await action(url);
+          await server.shutdown();
           resolve();
         } catch (err) {
+          await server.shutdown();
           reject(err);
         }
-        await server.shutdown();
       },
     }, (request) => {
       const url = new URL(request.url);

src/shell.ts

@@ -1,3 +1,4 @@
+import { CommandPipeReader } from "../mod.ts";
 import { KillSignal } from "./command.ts";
 import { CommandContext, CommandHandler } from "./command_handler.ts";
 import { getExecutableShebangFromPath, ShebangInfo } from "./common.ts";
@@ -200,7 +201,7 @@ function cloneEnv(env: Env) {
 }
 
 export class Context {
-  stdin: ShellPipeReader;
+  stdin: CommandPipeReader;
   stdout: ShellPipeWriter;
   stderr: ShellPipeWriter;
   #env: Env;
@@ -209,7 +210,7 @@ export class Context {
   #signal: KillSignal;
 
   constructor(opts: {
-    stdin: ShellPipeReader;
+    stdin: CommandPipeReader;
     stdout: ShellPipeWriter;
     stderr: ShellPipeWriter;
     env: Env;
@@ -226,6 +227,12 @@ export class Context {
     this.#signal = opts.signal;
   }
 
+  [Symbol.dispose]() {
+    if (this.stdin instanceof ReadableStream) {
+      this.stdin.cancel();
+    }
+  }
+
   get signal() {
     return this.#signal;
   }
@@ -351,7 +358,7 @@ export function parseCommand(command: string) {
 }
 
 export interface SpawnOpts {
-  stdin: ShellPipeReader;
+  stdin: CommandPipeReader;
   stdout: ShellPipeWriter;
   stderr: ShellPipeWriter;
   env: Record<string, string>;
@@ -364,7 +371,7 @@ export interface SpawnOpts {
 export async function spawn(list: SequentialList, opts: SpawnOpts) {
   const env = opts.exportEnv ? new RealEnv() : new ShellEnv();
   initializeEnv(env, opts);
-  const context = new Context({
+  using context = new Context({
     env,
     commands: opts.commands,
     stdin: opts.stdin,
@@ -649,7 +656,7 @@ async function executeCommandArgs(commandArgs: string[], context: Context): Prom
     await stdinPromise;
   }
 
-  async function writeStdin(stdin: ShellPipeReader, p: Deno.ChildProcess, signal: AbortSignal) {
+  async function writeStdin(stdin: CommandPipeReader, p: Deno.ChildProcess, signal: AbortSignal) {
     if (typeof stdin === "string") {
       return;
     }