Merge pull request #14 from jupyterlite/pipes

Implement pipes

src/commands/builtin_command_runner.ts

@@ -22,7 +22,7 @@ export class BuiltinCommandRunner implements ICommandRunner {
     } else if (args.length > 1) {
       throw new Error("cd: too many arguments")
     }
-    
+
     let path = args[0]
     if (path == "-") {
       const oldPwd = context.environment.get("OLDPWD")

src/commands/coreutils_command_runner.ts

@@ -4,15 +4,10 @@ import { WasmCommandRunner } from "./wasm_command_runner"
 export class CoreutilsCommandRunner extends WasmCommandRunner {
   names(): string[] {
     return [
-      // File commands
-      "cp", "echo", "env", "ln", "ls", "mkdir", "mv", "pwd", "realpath", "rm", "rmdir", "touch",
-      "uname",
-      // Text commands
-      "cat", "cut", "head", "join", "md5sum", "nl", "sha1sum", "sha224sum", "sha256sum",
-      "sha384sum", "sha512sum", "sort", "tail", "tr", "wc",
-      // Shell commands
-      "basename", "date", "dirname", "echo", "env", "expr", "id", "logname", "pwd", "seq", "sleep",
-      "stat", "uname",
+      "basename", "cat", "cp", "cut", "date", "dirname", "echo", "env", "expr", "head", "id",
+      "join", "ln", "logname", "ls", "md5sum", "mkdir", "mv", "nl", "pwd", "realpath", "rm",
+      "rmdir", "seq", "sha1sum", "sha224sum", "sha256sum", "sha384sum", "sha512sum", "sleep",
+      "sort", "stat", "tail", "touch", "tr", "uname", "uniq", "wc",
     ]
   }
 

src/context.ts

@@ -1,6 +1,6 @@
 import { Environment } from "./environment"
 import { IFileSystem } from "./file_system"
-import { Input, Output } from "./io"
+import { IInput, IOutput } from "./io"
 
 /**
  * Context used to run commands.
@@ -11,8 +11,8 @@ export class Context {
     readonly fileSystem: IFileSystem,
     readonly mountpoint: string,
     environment: Environment,
-    readonly stdin: Input,
-    readonly stdout: Output,
+    readonly stdin: IInput,
+    readonly stdout: IOutput,
   ) {
     this.environment = environment
   }

src/io/buffered_output.ts

@@ -1,15 +1,17 @@
-import { Output } from "./output"
+import { IOutput } from "./output"
 
-export abstract class BufferedOutput extends Output {
-  constructor() {
-    super()
+export abstract class BufferedOutput implements IOutput {
+  protected get allContent(): string {
+    return this.data.join("")
   }
 
   protected clear() {
     this.data = []
   }
 
-  override async write(text: string): Promise<void> {
+  abstract flush(): Promise<void>
+
+  async write(text: string): Promise<void> {
     this.data.push(text)
   }
 

src/io/console_output.ts

@@ -1,9 +1,9 @@
-import { Output } from "./output"
+import { IOutput } from "./output"
 
-export class ConsoleOutput extends Output {
-  override async flush(): Promise<void> {}
+export class ConsoleOutput implements IOutput {
+  async flush(): Promise<void> {}
 
-  override async write(text: string): Promise<void> {
+  async write(text: string): Promise<void> {
     console.log(text)
   }
 }

src/io/file_input.ts

@@ -1,10 +1,8 @@
-import { Input } from "./input"
 import { IFileSystem } from "../file_system"
+import { IInput } from "./input"
 
-export class FileInput extends Input {
-  constructor(readonly fileSystem: IFileSystem, readonly path: string) {
-    super()
-  }
+export class FileInput implements IInput {
+  constructor(readonly fileSystem: IFileSystem, readonly path: string) {}
 
   read(): string {
     const { FS } = this.fileSystem

src/io/file_output.ts

@@ -8,7 +8,7 @@ export class FileOutput extends BufferedOutput {
 
   override async flush(): Promise<void> {
     const { FS } = this.fileSystem
-    let content = this.data.join("")
+    let content = this.allContent
 
     if (this.append) {
       try {

src/io/input.ts

@@ -1,6 +1,6 @@
-export abstract class Input {
+export interface IInput {
   /**
    * Read and return the entire contents of this input.
    */
-  abstract read(): string
+  read(): string
 }

src/io/output.ts

@@ -1,4 +1,4 @@
-export abstract class Output {
-  abstract flush(): Promise<void>
-  abstract write(text: string): Promise<void>
+export interface IOutput {
+  flush(): Promise<void>
+  write(text: string): Promise<void>
 }

src/io/pipe.ts

@@ -1,14 +1,13 @@
-import { Output } from "./output"
-
-export class Pipe extends Output {
-  constructor() {
-    super()
-  }
+import { BufferedOutput } from "./buffered_output"
+import { IInput } from "./input"
 
+export class Pipe extends BufferedOutput implements IInput {
   override async flush(): Promise<void> {
   }
 
-  override async write(text: string): Promise<void> {
-
+  read(): string {
+    const ret = this.allContent
+    this.clear()
+    return ret
   }
 }

src/io/redirect_output.ts

@@ -1,8 +1,8 @@
-import { Output } from "./output"
 import { BufferedOutput } from "./buffered_output"
+import { IOutput } from "./output"
 
 export class RedirectOutput extends BufferedOutput {
-  constructor(target: Output) {
+  constructor(target: IOutput) {
     super()
     this.target = target
   }
@@ -15,5 +15,5 @@ export class RedirectOutput extends BufferedOutput {
     await this.target.flush()
   }
 
-  private readonly target: Output
+  private readonly target: IOutput
 }

src/io/single_char_input.ts

@@ -1,10 +1,10 @@
-import { Input } from "./input"
+import { IInput } from "./input"
 
 /**
  * Wrapper for an Input that reads a single character at a time.
  */
 export class SingleCharInput {
-  constructor(readonly input: Input) {}
+  constructor(readonly input: IInput) {}
 
   /**
    * Return a single character at a time.  Return ascii 4 (EOT) when nothing more left to read.

src/io/terminal_input.ts

@@ -1,6 +1,6 @@
-import { Input } from "./input"
+import { IInput } from "./input"
 
-export class TerminalInput extends Input {
+export class TerminalInput implements IInput {
   read(): string {
     return ""
   }

src/shell.ts

@@ -2,9 +2,9 @@ import { CommandRegistry } from "./command_registry"
 import { Context } from "./context"
 import { Environment } from "./environment"
 import { IFileSystem } from "./file_system"
-import { FileInput, FileOutput, Input, Output, TerminalInput, TerminalOutput } from "./io"
+import { FileInput, FileOutput, IInput, IOutput, Pipe, TerminalInput, TerminalOutput } from "./io"
 import { IOutputCallback } from "./output_callback"
-import { CommandNode, parse } from "./parse"
+import { CommandNode, PipeNode, parse } from "./parse"
 import * as FsModule from './wasm/fs'
 
 export class Shell {
@@ -105,45 +105,24 @@ export class Shell {
     const stdin = new TerminalInput()
     const stdout = new TerminalOutput(this._outputCallback)
     try {
-      const cmdNodes = parse(cmdText)
-      const ncmds = cmdNodes.length
-
-      for (let i = 0; i < ncmds; ++i) {
-        const command = cmdNodes[i] as CommandNode
-        const cmdName = command.name.value
-
-        const runner = CommandRegistry.instance().get(cmdName)
-        if (runner === null) {
-          // Give location of command in input?
-          throw new Error(`Unknown command: '${cmdName}'`)
-        }
-
-        let input: Input = stdin
-        let output: Output = stdout
-        if (command.redirects) {
-          // Support single redirect only, write or append to file.
-          if (command.redirects.length > 1) {
-            throw new Error("Only implemented a single redirect per command")
-          }
-          const redirect = command.redirects[0]
-          const redirectChars = redirect.token.value
-          const path = redirect.target.value
-          if (redirectChars == ">" || redirectChars == ">>") {
-            output = new FileOutput(this._fileSystem!, path, redirectChars == ">>")
-          } else if (redirectChars == "<") {
-            input = new FileInput(this._fileSystem!, path)
-          } else {
-            throw new Error("Unrecognised redirect " + redirectChars)
+      const nodes = parse(cmdText)
+
+      for (const node of nodes) {
+        if (node instanceof CommandNode) {
+          await this._runCommand(node, stdin, stdout)
+        } else if (node instanceof PipeNode) {
+          const { commands } = node
+          const n = commands.length
+          let prevPipe: Pipe
+          for (let i = 0; i < n; i++) {
+            const input = i == 0 ? stdin : prevPipe!
+            const output = i < n-1 ? (prevPipe = new Pipe()) : stdout
+            await this._runCommand(commands[i], input, output)
           }
+        } else {
+          // This should not occur.
+          throw new Error(`Expected CommandNode or PipeNode not ${node}`)
         }
-
-        const args = command.suffix.map((token) => token.value)
-        const context = new Context(
-          args, this._fileSystem!, this._mountpoint, this._environment, input, output,
-        )
-        await runner.run(cmdName, context)
-
-        await context.flush()
       }
     } catch (error: any) {
       // Send result via output??????  With color.  Should be to stderr.
@@ -152,6 +131,44 @@ export class Shell {
     }
   }
 
+  private async _runCommand(
+    commandNode: CommandNode,
+    input: IInput,
+    output: IOutput,
+): Promise<void> {
+    const name = commandNode.name.value
+    const runner = CommandRegistry.instance().get(name)
+    if (runner === null) {
+      // Give location of command in input?
+      throw new Error(`Unknown command: '${name}'`)
+    }
+
+    if (commandNode.redirects) {
+      // Support single redirect only, write or append to file.
+      if (commandNode.redirects.length > 1) {
+        throw new Error("Only implemented a single redirect per command")
+      }
+      const redirect = commandNode.redirects[0]
+      const redirectChars = redirect.token.value
+      const path = redirect.target.value
+      if (redirectChars == ">" || redirectChars == ">>") {
+        output = new FileOutput(this._fileSystem!, path, redirectChars == ">>")
+      } else if (redirectChars == "<") {
+        input = new FileInput(this._fileSystem!, path)
+      } else {
+        throw new Error("Unrecognised redirect " + redirectChars)
+      }
+    }
+
+    const args = commandNode.suffix.map((token) => token.value)
+    const context = new Context(
+      args, this._fileSystem!, this._mountpoint, this._environment, input, output,
+    )
+    await runner.run(name, context)
+
+    await context.flush()
+  }
+
   private async _tabComplete(text: string): Promise<[number, string[]]> {
     // Assume tab completing command.
     return [text.length, CommandRegistry.instance().match(text)]

tests/shell.test.ts

@@ -30,6 +30,16 @@ describe("Shell", () => {
       await shell._runCommands("wc < file2")
       expect(output.text).toEqual("      1       5      27\r\n")
     })
+
+    it("should support pipe", async () => {
+      const { shell, output } = await shell_setup_simple()
+      await shell._runCommands("ls -1|sort -r")
+      expect(output.text).toEqual("file2\r\nfile1\r\ndirA\r\n")
+      output.clear()
+
+      await shell._runCommands("ls -1|sort -r|uniq -c")
+      expect(output.text).toEqual("      1 file2\r\n      1 file1\r\n      1 dirA\r\n")
+    })
   })
 
   describe("input", () => {