Repository F# setup

open System
open System.IO
open System.Threading
open System.Threading.Tasks
open Axial
open Axial.Console
open Axial.FileSystem
open Axial.PlatformService
open Axial.Process

Streaming output

streamStdoutLines yields decoded stdout lines as they arrive. It returns a ProcessStream, and its terminal returns the consumed value together with process completion. Stderr stays separate unless you apply mergeStderr.

ProcessStream<'env, 'value> is the stream type; the ProcessStream module provides ProcessStream.map, filter, take, and the run* terminals. A stream terminal returns a Flow, which a script executes with Process.run or an application binds with let! inside the enclosing Flow. A ProcessStream cannot be passed to Process.run directly.

Show page setup

// Setup for the checked examples on this page.
open Axial.Console
open Axial.Process
open Axial.Process.DSL
/// Fails the docs test when an example's result differs from the value shown.
let shouldEqual expected actual =
    if actual <> expected then failwithf "Expected %A but got %A" expected actual

Verified

let lines =
    cmd $"printf 'ready\nsample 1\n'"
    |> streamStdoutLines

let reports = lines |> ProcessStream.map _.ToUpperInvariant() |> ProcessStream.runCollect

lines: ProcessStream<ProcessHostEnvironment,string>

cmd: FormattableString -> ProcessCommand

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

streamStdoutLines: ^a -> ProcessStream<'b,string>

reports: Flow<ProcessHostEnvironment,ProcessError,(string list * ProcessCompletion)>

Axial.Process.ProcessStream

map: ('a -> 'b) -> ProcessStream<'c,'a> -> ProcessStream<'c,'b>

Transforms values while preserving process completion.

_arg1: string

ToUpperInvariant: unit -> string

Returns a copy of this object converted to uppercase using the casing rules of the invariant culture. The uppercase equivalent of the current string.

runCollect: ProcessStream<'a,'b> -> Flow<'a,ProcessError,('b list * ProcessCompletion)>

Collects every selected value plus completion. The collected list can grow without bound.

Requires & types

Requires: open Axial.Process; open Axial.Process.DSL (packages: Axial.Process)

val lines: ProcessStream<ProcessHostEnvironment,string>

val reports: Flow<ProcessHostEnvironment,ProcessError,(string list * ProcessCompletion)>

Verified

let values, completion = reports |> Process.run
shouldEqual ["READY"; "SAMPLE 1"] values
match completion with
| ProcessCompletion.Completed result -> shouldEqual 0 result.ExitCode
| ProcessCompletion.StoppedEarly -> failwith "Expected completed output"

values: string list

completion: ProcessCompletion

reports: Flow<ProcessHostEnvironment,ProcessError,(string list * ProcessCompletion)>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

Axial.Process.Process

run: ^source -> ^result

Runs with live services and returns the value; raises ProcessRunException on failure.

shouldEqual: 'a -> 'a -> unit

Axial.Process.ProcessCompletion

Completed

result: ProcessResult

ExitCode: int

StoppedEarly

failwith: string -> 'T

Throw a exception. The exception message. Never returns. let failingFunction() = failwith "Oh no" // Throws an exception true // Never reaches this failingFunction() // Throws a System.Exception

Requires & types

Requires: open Axial.Process (packages: Axial.Process)

val values: string list

val completion: ProcessCompletion

lines is a ProcessStream<'env, string>: the value parameter is one decoded line, and 'env supplies process and clock services when it runs. runCollect produces Flow<'env, ProcessError, string list * ProcessCompletion>; Process.run returns the values and completion directly. Completed result includes the final ProcessResult; StoppedEarly reports successful prefix stopping after cleanup.

map and filter preserve completion. streamStdout and streamStderr emit decoded chunks with stage, timestamp, ObservedChannel, and OriginalChannel metadata. streamOutputs includes all readable, unconnected channels. A merged stderr value has observed channel stdout and original channel stderr.

Handle output with application effects

Use runForEachFlow when a line needs a console, filesystem, or other application service. runForEach takes a synchronous callback and suits ordinary in-memory work. Effectful terminals keep ProcessFailed and ConsumerFailed distinct; defects and interruption retain their Axial cause. Their Flow error type is ProcessStreamError<'consumerError>; use Process.tryRun to inspect one inside Exit.

Verified

let printLines =
    lines
    |> ProcessStream.runForEachFlow (Console.writeLine<_, string>)

printLines: Flow<ProcessHostEnvironment,ProcessStreamError<string>,ProcessCompletion>

lines: ProcessStream<ProcessHostEnvironment,string>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

Axial.Process.ProcessStream

runForEachFlow: ('a -> Flow<'b,'c,unit>) -> ProcessStream<'b,'a> -> Flow<'b,ProcessStreamError<'c>,ProcessCompletion>

Runs each callback in the application environment; preserves consumer errors separately.

Axial.Console.ConsoleModule

writeLine: string -> Flow<'env,'error,unit>

string

An abbreviation for the CLI type . Basic Types

Requires & types

Requires: open Axial.Console; open Axial.Process (packages: Axial.Console, Axial.Process)

val printLines: Flow<ProcessHostEnvironment,ProcessStreamError<string>,ProcessCompletion>

Verified

match printLines |> Process.run with
| ProcessCompletion.Completed result -> shouldEqual 0 result.ExitCode
| ProcessCompletion.StoppedEarly -> failwith "Expected completed output"

printLines: Flow<ProcessHostEnvironment,ProcessStreamError<string>,ProcessCompletion>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

Axial.Process.Process

run: ^source -> ^result

Runs with live services and returns the value; raises ProcessRunException on failure.

Axial.Process.ProcessCompletion

Completed

result: ProcessResult

shouldEqual: 'a -> 'a -> unit

ExitCode: int

StoppedEarly

failwith: string -> 'T

Throw a exception. The exception message. Never returns. let failingFunction() = failwith "Oh no" // Throws an exception true // Never reaches this failingFunction() // Throws a System.Exception

Requires & types

Requires: open Axial.Process (packages: Axial.Process)

Stop after a prefix

take and takeWhile stop and wait for the owned native process to clean up before returning a successful StoppedEarly. take 0 never launches it. Caller cancellation still interrupts the Flow.

Verified

let firstLine =
    cmd $"sh -c 'echo ready; sleep 30'"
    |> streamStdoutLines
    |> ProcessStream.take 1
    |> ProcessStream.runCollect

firstLine: Flow<ProcessHostEnvironment,ProcessError,(string list * ProcessCompletion)>

cmd: FormattableString -> ProcessCommand

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

streamStdoutLines: ^a -> ProcessStream<'b,string>

Axial.Process.ProcessStream

take: int -> ProcessStream<'a,'b> -> ProcessStream<'a,'b>

Stops the owned process after a prefix. Zero does not start a process.

runCollect: ProcessStream<'a,'b> -> Flow<'a,ProcessError,('b list * ProcessCompletion)>

Collects every selected value plus completion. The collected list can grow without bound.

Requires & types

Requires: open Axial.Process; open Axial.Process.DSL (packages: Axial.Process)

val firstLine: Flow<ProcessHostEnvironment,ProcessError,(string list * ProcessCompletion)>

Verified

firstLine
|> Process.run
|> shouldEqual (["ready"], ProcessCompletion.StoppedEarly)

firstLine: Flow<ProcessHostEnvironment,ProcessError,(string list * ProcessCompletion)>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

Axial.Process.Process

run: ^source -> ^result

Runs with live services and returns the value; raises ProcessRunException on failure.

shouldEqual: 'a -> 'a -> unit

Axial.Process.ProcessCompletion

StoppedEarly

Requires & types

Requires: open Axial.Process (packages: Axial.Process)

Bound retained output

Streams keep a 64 KiB stdout tail and separate 64 KiB stderr tails by default. Backpressure bounds pending output, but runCollect still retains every value it collects, so consume unbounded output incrementally with runForEachFlow or runFoldFlow. captureAll opts into full stdout retention; captureNothing removes stdout retention while keeping observations and the separate stderr diagnostic tail.

Verified

let retainedTail =
    cmd $"printf 'first\nlast\n'"
    |> captureTail 5
    |> streamStdoutLines
    |> ProcessStream.runDrain

retainedTail: Flow<ProcessHostEnvironment,ProcessError,ProcessCompletion>

cmd: FormattableString -> ProcessCommand

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

captureTail: int -> ^a -> 'b

Same as Process.captureTail. Bounds final stdout retention in bytes.

streamStdoutLines: ^a -> ProcessStream<'b,string>

Axial.Process.ProcessStream

runDrain: ProcessStream<'a,'b> -> Flow<'a,ProcessError,ProcessCompletion>

Consumes selected output without accumulating values and returns completion.

Requires & types

Requires: open Axial.Process; open Axial.Process.DSL (packages: Axial.Process)

val retainedTail: Flow<ProcessHostEnvironment,ProcessError,ProcessCompletion>

Verified

match retainedTail |> Process.run with
| ProcessCompletion.Completed result ->
    shouldEqual "last\n" result.StdOut
    shouldEqual true result.StdOutTruncated
| ProcessCompletion.StoppedEarly -> failwith "Expected completed output"

retainedTail: Flow<ProcessHostEnvironment,ProcessError,ProcessCompletion>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

Axial.Process.Process

run: ^source -> ^result

Runs with live services and returns the value; raises ProcessRunException on failure.

Axial.Process.ProcessCompletion

Completed

result: ProcessResult

shouldEqual: 'a -> 'a -> unit

StdOut: string

StdOutTruncated: bool

StoppedEarly

failwith: string -> 'T

Throw a exception. The exception message. Never returns. let failingFunction() = failwith "Oh no" // Throws an exception true // Never reaches this failingFunction() // Throws a System.Exception

Requires & types

Requires: open Axial.Process (packages: Axial.Process)

Use ProcessStream.toFlowStream when you need an ordinary FlowStream API; merging process output with other sources, feeding queues and hubs, folding, stderr provenance, line limits, and conversion are covered in the FlowStream integration recipe.

Streaming output

Handle output with application effects

Stop after a prefix

Bound retained output

Overview

Docs

API Reference