Pipelines and input

import { run } from "@j50n/proc";

// cut -d " " -f 3 app.log | sort | uniq -c
const counts = await run("cut", "-d", " ", "-f", "3", "app.log")
  .run("sort")
  .run("uniq", "-c")
  .lines
  .map((line) => line.trim())
  .collect();

console.log(counts);

[ "2 ERROR", "4 INFO", "1 WARN" ]

.run() on a command’s output starts another command with that output as its stdin, as | does in a shell. Every command in the chain runs at the same time, and the bytes pass from one to the next in chunks; nothing is collected in between unless you collect it. If any command fails, the consumer at the end throws; see Errors.

Feeding a command from an array

import { enumerate } from "@j50n/proc";

const sorted = await enumerate(["pear", "apple", "fig"])
  .run("sort")
  .lines
  .collect();
console.log(sorted);

// Only text and bytes can be written; turn anything else into strings first.
const numbers = await enumerate([10, 9, 100])
  .map((n) => `${n}`)
  .run("sort", "-n")
  .lines
  .collect();
console.log(numbers);

[ "apple", "fig", "pear" ]
[ "9", "10", "100" ]

.run() works on any Enumerable, not only on a command’s output. enumerate() wraps an array, a Set, a generator, or any other iterable or async iterable, and .run() writes its items to the command’s stdin.

Feeding a command from a file

import { read } from "@j50n/proc";

// grep -c ERROR < app.log
const count = await read("app.log").run("grep", "-c", "ERROR").lines.first;

console.log(count);

2

read() yields a file’s bytes, and .run() sends them to the command unchanged. It is the same as grep -c ERROR < app.log, without an extra cat process. The file is opened when reading starts, which is at the .run() call. If it is missing, the command sees empty input, and the consumer throws with Deno.errors.NotFound as the cause: an UpstreamError, or the command’s own ExitCodeError if empty input makes it fail, as it does grep -c (which prints 0 and exits 1).

How items become stdin

import { concat, enumerate } from "@j50n/proc";

const decoder = new TextDecoder();
const encoder = new TextEncoder();

// `cat` echoes its stdin, so this shows exactly what the command received.
async function received(
  items: Iterable<string | string[] | Uint8Array | Uint8Array[]>,
) {
  const bytes = concat(await enumerate(items).run("cat").collect());
  return JSON.stringify(decoder.decode(bytes));
}

console.log(await received(["one", "two"])); // each string is a line
console.log(await received([["one", "two"], ["three"]])); // so is each element
console.log(await received([encoder.encode("raw"), encoder.encode("bytes")]));

"one\ntwo\n"
"one\ntwo\nthree\n"
"rawbytes"

So lines from .lines go back in as lines, and bytes from a command or a file go through untouched. For any other item type (numbers, objects), the return type of .run() is never and the code doesn’t type-check: turn the items into strings with .map() first, as the array example does with numbers.

Lots of lines

Each item written to a command is a write of its own, and each step pays for every item it handles. With a few thousand lines that doesn’t matter; with millions it does. Work a chunk of lines at a time instead:

import { run } from "@j50n/proc";

// Lines arrive in arrays, one per chunk read. A step works on the whole
// array, and .run() writes each array to the next command in one go.
const count = await run("seq", "1", "1000000")
  .chunkedLines
  .map((lines) => lines.filter((line) => !line.endsWith("7")))
  .run("wc", "-l")
  .lines
  .first;

console.log(count);

900000

.chunkedLines yields the same lines as .lines, in arrays, and .run() writes each array in one go. On two million short lines, a filter between two commands took 16 seconds a line at a time and 0.3 seconds a chunk at a time.

run({ buffer: true }, ...) is a smaller fix for a source you can’t chunk: it collects small items into writes of at least 16 KiB, about twice as fast. The child sees nothing until 16 KiB has collected or the input ends, so leave it off for a child that must answer each line as it arrives.

Mixing commands and steps

import { run } from "@j50n/proc";

const messages = await run("cat", "app.log")
  .lines
  .filter((line) => line.includes(" ERROR "))
  .map((line) => line.slice(26)) // drop the timestamp and level
  .run("sort", "-f") // -f: ignore case
  .lines
  .collect();

console.log(messages);

[ "db: connection refused", "GET /orders/17 500" ]

Any step can sit between two commands: .lines turns the bytes into text, .filter() and .map() work on each line, and the next .run() writes the strings back out as lines. Use a command where a tool already does the job well (sort on data larger than memory, grep on a big file), and a TypeScript step where the logic is easier to write and test in code.

Stopping early

import { run } from "@j50n/proc";

// The consumer stops: take(2) closes the pipeline behind it.
const sevens = await run("seq", "1", "1000000")
  .run("grep", "7")
  .lines
  .take(2)
  .collect();
console.log(sevens);

// A command stops: head exits after two lines, and `yes` stops with it.
const ys = await run("yes").run("head", "-n", "2").lines.collect();
console.log(ys);

[ "7", "17" ]
[ "y", "y" ]

There are two ways a pipeline stops before its first command runs out, and neither is an error:

• The consumer stops (take, first, find, a break out of for await). proc closes the pipeline behind it, and the await returns at once. seq and grep die of SIGPIPE at their next write. Their exit codes are not checked.
• A command stops reading, like head. Writing to it ends quietly, and the command before it is closed the same way.

A command that goes quiet instead keeps running after the await has returned, until it writes again, exits, or is stopped. That is what lets you start a server and wait for its “ready” line. Deno doesn’t exit while a child is still running, though, so without main() the script ends only when the server does; under main(), the server is stopped when the program ends.

When a pipeline is cut short, the commands before the cut may or may not have finished, so don’t rely on their failures being reported. A command whose output proc read to the end has its exit code checked; one that was closed early doesn’t. With head in the middle, which of the two happens depends on timing: run("sh", "-c", "echo a; echo b; exit 3").run("head", "-n", "1") throws an UpstreamError for the exit 3 on some runs, and returns ["a"] without one on others.

Pipelines run for their side effects

import { read, run } from "@j50n/proc";

// Into a file: writeTo is the consumer.
await read("app.log").run("grep", "WARN").writeTo("warnings.log");
console.log((await Deno.readTextFile("warnings.log")).trimEnd());

// A command run only for what it does: consume its (empty) output, so a
// failure throws.
await run("cp", "warnings.log", "warnings.bak").forEach(() => {});
console.log((await Deno.stat("warnings.bak")).isFile);

2026-10-05 09:01:14 WARN  retrying db connection
true

A pipeline that writes a file ends in .writeTo(path), which is its consumer; it takes bytes, which is what a command produces. A command you run only for what it does (cp, mkdir, git commit) still needs a consumer, and .forEach(() => {}) is the one to use: it reads and discards any output, and throws if the command fails. .toStdout() is the other common ending, when the output is for the user to see.

.run() starts at once

Like run(), .run() starts its command when you call it, and it begins reading the items before it into the command’s stdin straight away, whether or not anything reads the output yet. Callbacks in the steps before it (a .map() that logs, say) run then too. The steps after it are lazy, as usual. Build a pipeline only when you are about to consume it, and always consume it: an unread pipeline can hang (see Waiting for a command).

To write a child’s stdin a piece at a time, as your program produces the data, use Process and its stdin, or feed .run() from a WritableIterable.