API at a glance

Everything @j50n/proc and @j50n/proc/transforms export, grouped by what you would use it for, one line each. Each name links to its full entry on JSR. Key ideas explains how the pieces fit.

import { enumerate, read, run } from "@j50n/proc";
import { fromCsvToRows, toTsv } from "@j50n/proc/transforms";

Running commands

• run(...cmd): start a command now; returns its stdout to read. Options go first: run({ cwd, env }, "ls").
• .run(...cmd): pipe the items (lines or bytes) into a command’s stdin; returns its output.
• ProcessEnumerable: what run() returns; an Enumerable of stdout bytes, plus .pid and .status.
• .status: a promise of the exit status, without throwing on failure. A getter.
• .pid: the child’s process ID. A getter.
• ProcessOptions: cwd, env, fnStderr, fnError, buffer.
• StderrHandler: the type of fnStderr, which reads the child’s stderr instead of letting it reach the terminal.
• ErrorHandler: the type of fnError, which decides what a failure throws, or suppresses it.
• Cmd: a command and its arguments, [program, ...args].
• Process: the low-level child process under run(); use it to write stdin as you go, or to choose how each stream is connected.
• ProcessStreamOptions: options for new Process: ProcessOptions plus stdin, stdout, stderr.
• PipeKinds: "piped", "inherit", or "null".

Errors

• ProcessError: base class of the four below; catch it to handle any process failure.
• ExitCodeError: a command exited non-zero; .code, .command.
• SignalError: a command was killed by a signal; .signal, .command.
• TimeoutError: a command ran past its timeoutMs and was stopped; .timeoutMs, .command.
• UpstreamError: a command succeeded but its input failed; .cause is the original error.

Shutting down

• main(program, { timeoutMs }): run your program, and however it ends, stop the children and wait for them before exiting. For containers and services.
• terminateAll({ signal, timeoutMs }): signal every child proc started and wait for them, without exiting.

Sources

• enumerate(iterable): wrap an array, Set, generator, stream, or any (async) iterable as an Enumerable.
• read(path): a file’s bytes, opened when reading starts.
• readLines(path): a UTF-8 file’s lines; the same as read(path).lines.
• range({ from, to | until, step }): a sequence of numbers.
• RangeToOptions, RangeUntilOptions: the two shapes of range’s options (to excludes the end, until includes it).
• WritableIterable: a queue you write() to from callbacks and read with for await. No backpressure: writes never wait.
• Writable: something with write() and close(): a WritableIterable or a process’s stdin.

Enumerable

Enumerable is the async sequence every source returns. Steps return a new Enumerable; consumers return a promise. Read it once.

Steps:

• map(fn): transform each item; fn may be async.
• filter(fn), filterNot(fn): keep, or drop, the items fn accepts.
• flatMap(fn): map each item to an iterable and yield its items.
• flatten(): yield the items of each item; turns parser batches into rows.
• enum(): number the items as [item, index].
• concurrentMap(fn, { concurrency }): map with several calls at once; results in input order.
• concurrentUnorderedMap(fn, { concurrency }): map with several calls at once; results as they finish.
• ConcurrentOptions: concurrency, default navigator.hardwareConcurrency.
• transform(fn | stream): pass the whole sequence through a transformer function or a TransformStream, such as DecompressionStream or fromCsvToRows().
• take(n): the first n items, then close the source.
• drop(n): skip the first n items.
• concat(other): these items, then other’s.
• zip(other): pair items by position.
• unzip(): split pairs into two Enumerables (holds items in memory, as tee does).
• tee(n): split into n copies that each see every item; keeps items in memory until all have read them.
• .lines: decode bytes into lines of text. A getter.
• .chunkedLines: the same lines in arrays, one per chunk; faster for many short lines. A getter.
• run(...cmd): pipe into a command (see above). Starts at the call, unlike the other steps.

Consumers:

• collect(), toArray(): every item, in an array.
• forEach(fn): call fn on each item, waiting for each.
• reduce(fn, zero): fold the items into one value.
• count(fn?): how many items, or how many pass fn.
• find(fn): the first item fn accepts, or undefined.
• some(fn), every(fn): whether any, or all, items pass fn.
• .first: a promise of the first item; RangeError if there is none. A getter.
• writeTo(path | stream | writable): write to a file (bytes, or strings as lines; { atomic: true } replaces it only once everything is written), a WritableStream, or a Writable.
• writeBytesTo(writer): write bytes to a Writer & Closer such as a Deno.FsFile, then close it.
• toStdout(): write lines or bytes to stdout.
• for await (const item of e): iterate it yourself.

Transformers

Functions to pass to .transform().

• toLines, toChunkedLines: bytes to lines, or arrays of lines; what .lines and .chunkedLines use.
• toByteLines: split bytes into lines without decoding, for data that isn’t UTF-8.
• toBytes: lines of text (or bytes) to byte chunks, a newline after each string; use before writeTo or a CompressionStream.
• buffer(size): join small byte chunks into chunks of at least size bytes.
• gzip, gunzip: compress and decompress; for plain bytes, CompressionStream and DecompressionStream do the same.
• jsonParse: parse each line as JSON; a blank line throws.
• jsonStringify: each item to a line of JSON.
• debug: log each item as it passes (to stdout), unchanged.
• transformerFromTransformStream(stream): wrap a TransformStream as a transformer function.
• TransformerFunction: the type of a transformer, (AsyncIterable<T>) => AsyncIterable<U>; an async function* is the usual way to write one.
• TransformStream: the { writable, readable } pair .transform() also accepts.
• StandardData: what toBytes and a process’s stdin accept: string, string[], Uint8Array, Uint8Array[].

toBufferSource is deprecated; use toBytes.

Utilities

• sleep(ms): wait.
• SECONDS, MINUTES, HOURS, DAYS, WEEKS: milliseconds, for sleep(2 * SECONDS) or a cache timeout.
• cache(key, value, { timeout, refresh }): keep a computed value in Deno KV across runs; needs --unstable-kv.
• fetchRecord(key): read the raw cache entry, expired or not, for debugging.
• concat(arrays): join byte arrays into one.
• concatLines(arrays): join byte arrays with a newline after each.
• writeAll(data, writer): write all the bytes to a Writer, such as Deno.stdout.
• shuffle(array): shuffle in place.
• isString(value): type guard for a string primitive.

Helper types

These name the result types of some methods; you rarely write them.

• Lines, ChunkedLines, ByteSink, Run: what .lines, .chunkedLines, writeBytesTo, and .run() return; never when the items are the wrong type, so the mistake fails to type-check.
• ElementType: the item type of an iterable; what flatten() yields.
• Unzip: what unzip() returns.
• Tuple, TupleOf: what tee(n) returns.

Data formats: @j50n/proc/transforms

Parsers take bytes and yield batches (arrays of rows); add .flatten() for one row at a time. Writers take rows or batches and yield bytes. See Data formats.

• fromCsvToRows(options): parse CSV into batches of string[] rows.
• fromCsvToLazyRows(options): parse CSV into batches of LazyRows, which decode a field only when read.
• toCsv(options): write rows as CSV, quoting where needed.
• csvToTsv(options), tsvToCsv(options): convert CSV to TSV and back, bytes to bytes, without making rows.
• CsvParseOptions, CsvStringifyOptions: separator, and crlf for writing.
• fromTsvToRows(), fromTsvToLazyRows(): parse TSV.
• toTsv(): write TSV; throws on a tab, CR, or LF in a field.
• fromJsonToRows(options): parse JSON lines into batches of values, optionally checked by a schema.
• toJson(): write one value per item as JSON lines; throws a TypeError on an item with no JSON form.
• JsonOptions: schema and sampleSize.
• ZodSchema: anything with a parse(value) method, a Zod schema included.
• fromRecordToRows(), fromRecordToLazyRows(): parse the record format.
• toRecord(): write the record format, for handing rows to another program.
• FIELD_SEPARATOR, RECORD_SEPARATOR: "\x1F" and "\x1E", the record format’s separators.
• Row: one row, a string[].
• LazyRow: a row that decodes fields on demand: getField(i), fieldEquals(i, value), columnCount, toStringArray(), and LazyRow.fromStringArray(fields) to make one.
• BATCH_SIZE_BYTES: about 128 KiB, how much input makes a batch.

The package also has a command-line converter, jsr:@j50n/proc/flatdata; see The flatdata CLI.