fix(runtime): make readlines aboartable (#65)

Author: c4spar

src/runtime/deps.ts

@@ -13,6 +13,7 @@ export type {
   ParsedPath,
 } from "https://deno.land/[email protected]/path/mod.ts";
 export * as io from "https://deno.land/[email protected]/io/mod.ts";
+export { BufReader } from "https://deno.land/[email protected]/io/buffer.ts";
 export type {
   ReadableStreamFromReaderOptions as IOReadableStreamFromReaderOptions,
   ReadLineResult,
@@ -45,3 +46,4 @@ export type {
 } from "https://deno.land/[email protected]/flags/mod.ts";
 export { colors } from "https://deno.land/x/[email protected]/ansi/colors.ts";
 export { default as shq } from "https://esm.sh/[email protected]";
+export { concat } from "https://deno.land/[email protected]/bytes/mod.ts";

src/runtime/lib/readline.ts

@@ -0,0 +1,24 @@
+import { BufReader, concat } from "../deps.ts";
+
+export async function* readLines(
+  reader: Deno.Reader,
+  isCanceld: () => boolean,
+): AsyncIterableIterator<string> {
+  const bufReader = new BufReader(reader);
+  let chunks: Uint8Array[] = [];
+  const decoder = new TextDecoder();
+  while (!isCanceld()) {
+    const res = await bufReader.readLine();
+    if (!res) {
+      if (chunks.length > 0) {
+        yield decoder.decode(concat(...chunks));
+      }
+      break;
+    }
+    chunks.push(res.line);
+    if (!res.more) {
+      yield decoder.decode(concat(...chunks));
+      chunks = [];
+    }
+  }
+}

src/runtime/process.ts

@@ -1,6 +1,7 @@
 /// <reference path="../../types.d.ts" />
 
 import { Deferred, deferred } from "./deps.ts";
+import { readLines } from "./lib/readline.ts";
 import { ProcessError } from "./process_error.ts";
 import { ProcessOutput } from "./process_output.ts";
 
@@ -21,6 +22,7 @@ export class Process implements Promise<ProcessOutput> {
   #maxRetries = 0;
   #retries = 0;
   #throwErrors = true;
+  #isKilled = false;
 
   constructor(cmd: string, { errorContext }: ProcessOptions = {}) {
     this.#cmd = cmd;
@@ -66,6 +68,7 @@ export class Process implements Promise<ProcessOutput> {
   }
 
   kill(signo: Deno.Signal): void {
+    this.#isKilled = true;
     this.#process.kill(signo);
   }
 
@@ -117,9 +120,19 @@ export class Process implements Promise<ProcessOutput> {
       const [status] = await Promise.all([
         this.#process.status(),
         this.#process.stdout &&
-        read(this.#process.stdout, [stdout, combined], Deno.stdout),
+        read(
+          this.#process.stdout,
+          [stdout, combined],
+          Deno.stdout,
+          () => this.#isKilled,
+        ),
         this.#process.stderr &&
-        read(this.#process.stderr, [stderr, combined], Deno.stderr),
+        read(
+          this.#process.stderr,
+          [stderr, combined],
+          Deno.stderr,
+          () => this.#isKilled,
+        ),
       ]);
 
       let output = new ProcessOutput({
@@ -169,8 +182,9 @@ async function read(
   reader: Deno.Reader,
   results: Array<Array<string>>,
   outputStream: Deno.Writer,
+  isCanceld: () => boolean,
 ): Promise<Error | void> {
-  for await (const line of io.readLines(reader)) {
+  for await (const line of readLines(reader, isCanceld)) {
     for (const result of results) {
       result.push(line + "\n");
     }

test.ts

@@ -125,14 +125,18 @@ Deno.test({
     const start = Date.now();
     await assertRejects(
       async () => {
+        // @TODO: kill is not working with old bash and pipefail -euo.
+        const prefix = $.prefix;
+        $.prefix = "";
         $.shell = "/bin/bash";
         const child = $`sleep 10`;
-        child.kill("SIGKILL");
+        setTimeout(() => child.kill("SIGKILL"), 10);
         await child;
+        $.prefix = prefix;
       },
       ProcessError,
     );
-    assert(Date.now() - start < 100, "process.kill() took too long");
+    assert(Date.now() - start < 200, "process.kill() took too long");
   },
 });
 
@@ -144,12 +148,12 @@ Deno.test({
       async () => {
         $.shell = "/bin/zsh";
         const child = $`sleep 10`;
-        child.kill("SIGKILL");
+        setTimeout(() => child.kill("SIGKILL"), 10);
         await child;
       },
       ProcessError,
     );
-    assert(Date.now() - start < 100, "process.kill() took too long");
+    assert(Date.now() - start < 200, "process.kill() took too long");
   },
 });