Processes & Pipelines

Run shell commands, manage processes, and build pipelines

The system API provides a process execution layer for running shell commands, managing their lifecycle, and building pipelines. It is built around three core types: Command (a description of what to run), ProcessHandle (a live process), and Pipeline (composed execution).

Command

Command describes a single executable invocation with arguments, environment, and working directory. The constructor is internal — use the factory functions or the companion method instead.

Creating commands

// Simple command
val echo = command("echo", "hello")

// With environment variables
val cmd = command("cat", "/etc/passwd").env("LC_ALL", "C")

// With working directory
val ls = command("ls").cwd("/data")

// With stdin
val grep = command("grep", "pattern").stdin("hello world\npattern here\n")

// Shell script via sh -c
val shellCmd = Command.shell("ls /data && echo done")

Factory functions

// Required: program name or path
command(program: String): Command
command(program: String, vararg args: Any): Command

Command builder methods (return Command for chaining)

Method	Description
arg(a)	Add a single argument
args(vararg a)	Add multiple arguments
args(a: List<Any>)	Add a list of arguments
env(key, value)	Add an environment variable
env(map)	Add multiple environment variables
cwd(dir: File)	Set working directory
cwd(path: String)	Set working directory by path
stdin(source: Stdin)	Set stdin source
stdin(bytes: ByteArray)	Provide stdin as bytes
stdin(text: String)	Provide stdin as text
stdout(dest: Stdio)	Set stdout destination
stderr(dest: Stdio)	Set stderr destination

Stdin values

Value	Description
Stdin.Inherit	Inherit parent process stdin
Stdin.Pipe	Read from a pipe (default)
Stdin.Bytes(data)	Provide fixed byte array as stdin

Stdio values

Value	Description
Stdio.Inherit	Inherit parent process stdout/stderr
Stdio.Capture	Capture output (default)
Stdio.Null	Discard output (/dev/null)
Stdio.File(file)	Append output to a file

Command execution methods

Method	Returns	Description
output()	CommandResult	Execute and wait for completion, capture all output
spawn()	ProcessHandle	Start the process and get a handle
stream()	Flow<ProcessEvent>	Execute and receive real-time events
status()	Int	Execute and return only the exit code

Companion method

Command.shell(script: String): Command

Creates a Command that executes script via sh -c.

ProcessHandle

A ProcessHandle represents a running process. It is obtained from Command.spawn() or Command.pipeTo().

ProcessHandle properties

Property	Returns	Description
pid	Int	OS process ID
isRunning	Boolean	Whether the process is still running
stdin	OutputStream	Stream to write to the process’s stdin
stdout	InputStream	Stream to read the process’s stdout
stderr	InputStream	Stream to read the process’s stderr
exitCode	Int	Exit code (throws if still running)

ProcessHandle methods

Method	Returns	Description
waitFor()	CommandResult	Suspend until process exits, return full output
waitForTimeout(timeoutMillis: Long)	CommandResult?	Wait with timeout, return null if timed out
waitForTimeout(timeout: Duration)	CommandResult?	Wait with timeout using kotlin.time.Duration
flow()	Flow<ProcessEvent>	Real-time stream of stdout/stderr/exit events
kill()		Force-kill the process (SIGKILL)
terminate()		Gracefully terminate the process (SIGTERM)

Collecting output

val handle: ProcessHandle = cmd.spawn()

// Collect events as a flow
handle.flow().collect { event ->
    when (event) {
        is ProcessEvent.Stdout -> println(event.text)
        is ProcessEvent.Stderr -> System.err.println(event.text)
        is ProcessEvent.ExitCode -> println("Exit: ${event.code}")
    }
}

// Or wait for completion
val result: CommandResult = handle.waitFor()

Pipeline

Pipeline allows composing multiple processes where stdout of one feeds stdin of the next. The constructor is internal — build via Command.pipe() infix.

Creating a pipeline

// Two-command pipeline
val pipeline = command("echo", "hello") pipe command("wc", "-l")

// Multi-command pipeline
val pipeline = command("cat", "input.txt")
    .pipe(command("grep", "pattern"))
    .pipe(command("sort"))

Pipeline methods

Method	Returns	Description
execute()	CommandResult	Execute the pipeline and return the last command’s output
watch()	Flow<ProcessEvent>	Execute and receive real-time events from the last command
pipe(next: Command)	Pipeline	Append a command to the pipeline (infix)

Awaiting pipeline completion

val result: CommandResult = pipeline.execute()

Streaming extensions

The com.klyx.api.system package provides many extension functions for convenient process handling:

// Command extensions
cmd.outputText()              // Deprecated: use output().stdoutText
cmd.outputLines()             // Deprecated: use output().stdoutLines
cmd.isSuccess()               // True if exit code is 0
cmd.isFailure()               // True if exit code is non-zero
cmd.outputWithTimeout(timeout) // Execute with timeout
cmd.retry(times = 3)          // Retry on failure
cmd.result()                  // Execute and return Result<CommandResult>
cmd.pipeTo(destination)       // Pipe stdout to another command
cmd.combinedLines()           // Merge stdout+stderr as lines
cmd.stdoutBytes()             // Stream stdout as byte arrays
cmd.stderrBytes()             // Stream stderr as byte arrays
cmd.streamLines()             // Stream stdout as lines
cmd.streamErrLines()          // Stream stderr as lines

// ProcessHandle extensions
handle.waitForText()          // Wait and return stdout text
handle.waitForLines()         // Wait and return stdout lines
handle.waitForTimeoutText(timeout)  // Wait with timeout, return text
handle.waitForTimeoutLines(timeout) // Wait with timeout, return lines
handle.combinedLines()        // Merge stdout+stderr as lines
handle.stdoutBytes()          // Stream stdout as byte arrays
handle.stderrBytes()          // Stream stderr as byte arrays
handle.streamLines()          // Stream stdout as lines
handle.streamErrLines()       // Stream stderr as lines

// Flow extensions
flow.combinedLines()          // Merge stdout+stderr as lines
flow.stdoutBytes()            // Filter to stdout byte arrays
flow.stderrBytes()            // Filter to stderr byte arrays
flow.stdoutLines()            // Filter to stdout lines
flow.stderrLines()            // Filter to stderr lines

Process utilities

Extension functions on Process:

val pid: Int = process.pid()       // Get OS process ID
process.terminate()                // Send SIGTERM
process.kill()                     // Send SIGKILL

CommandResult

The result of a completed command execution:

Property	Type	Description
exitCode	Int	Process exit code
stdout	ByteArray	Raw stdout bytes
stderr	ByteArray	Raw stderr bytes
stdoutText	String	Stdout decoded as UTF-8
stderrText	String	Stderr decoded as UTF-8
stdoutLines	List<String>	Stdout split into lines
stderrLines	List<String>	Stderr split into lines

ProcessEvent

Events emitted during streaming execution:

Event	Description
ProcessEvent.Stdout(data)	Standard output data chunk (has .text property)
ProcessEvent.Stderr(data)	Standard error data chunk (has .text property)
ProcessEvent.ExitCode(code)	Process finished with exit code

Example

Run a command and collect its output:

val result = command("git", "log", "--oneline", "-5").output()
println(result.stdoutText)
println("Exit code: ${result.exitCode}")

Run a pipeline:

val pipeline = command("find", ".").pipe(command("grep", ".kt"))
val result = pipeline.execute()
println(result.stdoutLines)

Stream a process in real time:

val handle = command("tail", "-f", "logs.txt").spawn()
handle.flow().collect { event ->
    when (event) {
        is ProcessEvent.Stdout -> println(event.text)
        is ProcessEvent.ExitCode -> println("Done: ${event.code}")
        else -> {}
    }
}

Run with a timeout:

val result = command("sleep", "5").outputWithTimeout(2.seconds)
if (result == null) {
    println("Timed out!")
}