Process Pipeline

Source: tutorial/09_process_pipeline See in Playground

Reads a file, passes its content through the operating system’s sort command, and writes the sorted result. Mélodium orchestrates the subprocess and its I/O streams; the actual sorting is done entirely by sort itself.

Running

cd tutorial/09_process_pipeline
melodium run Compo.toml --input_file fruits.txt

This turns a file containing banana, apple, cherry, date, and elderberry into an alphabetically sorted sorted.txt.

Optional: add --api-report and an API token (MELODIUM_API_TOKEN) to see this run’s full trace on Cadence.CI .

How it works

There is no custom model here: spawnOnce is a convenience treatment that obtains a local executor on its own. The file’s bytes are connected straight into the subprocess’s stdin, and its stdout/stderr are treated like any other byte stream:

treatment main(
    const input_file: string = "fruits.txt",
    const output:      string = "sorted.txt"
) {
    startup()
 
    read: readLocal(path=input_file)
    startup.trigger -> read.trigger
 
    proc: spawnOnce(command=|command("sort", []))
    read.data       -> proc.stdin
    startup.trigger -> proc.launch
 
    procFailed: logErrorMessage(label="process", message="failed to spawn `sort`")
    procError:  logError(label="process")
    proc.failed -> procFailed.trigger
    proc.error  -> procError.message
 
    exitCode: unwrapOr<i32>(default=-1)
    logExitCode: logDataInfo<i32>(label="exit-code")
    logExit:     logInfoMessage(label="process", message="sort finished")
    proc.exit      -> exitCode.option,value -> logExitCode.data
    proc.completed -> logExit.trigger
 
    stderrText: decode()
    stderrLog:  logErrors(label="sort-stderr")
    proc.stderr     -> stderrText.data
    stderrText.text -> stderrLog.messages
 
    stdoutText: decode()
    proc.stdout -> stdoutText.data
 
    write:   writeTextLocal(path=output)
    logDone: logInfoMessage(label="process", message="sorted output written")
    stdoutText.text -> write.text
    write.finished  -> logDone.trigger
}

The data flow connects the file straight through the subprocess and back out to disk:

read.data (the file’s bytes) is connected straight into proc.stdin; sort starts reading as soon as bytes arrive, before the file has even finished being read. There is no “read the whole file, then start the subprocess” step, the subprocess is just another treatment with stream ports.

A subprocess is treated exactly like any other treatment with stream ports: stdin, stdout, and stderr are Stream<byte>, the same as an HTTP body or file content, so every idiom already used for decoding and encoding applies unchanged. |command(name, arguments) builds a Command value from a plain executable name and an argument vector, empty here.

proc.exit is a Block<Option<i32>> (none if the process was killed by a signal rather than exiting normally); unwrapOr supplies a fallback. Note that this is the block variant of unwrapOr (from std/ops/option/block), and not the stream variant used elsewhere in the tutorial: they do the same job on different port kinds, and picking the wrong one is a type error that melodium check catches immediately.

Both stdout and stderr are decoded and logged or written directly. A successful run with nothing on stderr produces no items on stderrText.text at all, so stderrLog logs nothing, not an empty line.

Dependencies

[dependencies]
std      = "0.10.3"  # core flows, logging, data structures
fs       = "0.10.3"  # local file I/O
process  = "0.10.3"  # external process execution
encoding = "0.10.3"  # UTF-8 encode / decode