Documentation
¶
Index ¶
- Constants
- func Accumulate[T any](fn func([]T) ([]T, error)) gloo.Command[T, T]
- func Aggregate[In, Out any](fn func([]In) (Out, error)) gloo.Command[In, Out]
- func Drop[T any](n Count) gloo.Command[T, T]
- func Expand[In, Out any](fn func(In) ([]Out, error)) gloo.Command[In, Out]
- func Fail[T any](err error) gloo.Command[T, T]
- func Filter[T any](fn func(T) (bool, error)) gloo.Command[T, T]
- func FilterMap[In, Out any](fn func(In) (Out, bool, error)) gloo.Command[In, Out]
- func Head[T any](n Count) gloo.Command[T, T]
- func Map[In, Out any](fn func(In) (Out, error)) gloo.Command[In, Out]
- func MergeSorted[T any](ctx context.Context, less Less[T], a, b gloo.Stream[T]) gloo.Stream[T]
- func Run(target any) error
- func Scan[In, Out any](factory func() (step func(In) ([]Out, error), end func() ([]Out, error))) gloo.Command[In, Out]
- func StatefulFilter[T any](factory func() func(T) (bool, error)) gloo.Command[T, T]
- func StatefulFilterMap[In, Out any](factory func() func(In) (Out, bool, error)) gloo.Command[In, Out]
- func StatefulMap[In, Out any](factory func() func(In) (Out, error)) gloo.Command[In, Out]
- func Subprocess(name ProcessName, args ...ProcessArg) gloo.Command[[]byte, []byte]
- func Take[T any](n Count) gloo.Command[T, T]
- func Tap[T any](fn func(T) error) gloo.Command[T, T]
- type Count
- type Less
- type ProcessArg
- type ProcessName
Examples ¶
Constants ¶
const ( // ErrRunNilTarget reports a nil target. ErrRunNilTarget gloo.Error = "patterns: run: target is nil" // ErrRunUnsupportedTarget reports a target that is neither a // Source[[]byte] nor a Command[[]byte, []byte]. ErrRunUnsupportedTarget gloo.Error = "patterns: run: target implements neither Source[[]byte] nor Command[[]byte, []byte]" )
Errors surfaced by Run when the target cannot be executed at all. Pipeline failures (stream errors, sink errors) are returned as-is.
const ( ErrSubprocessStdinPipe gloo.Error = "subprocess: stdin pipe" ErrSubprocessStdoutPipe gloo.Error = "subprocess: stdout pipe" ErrSubprocessStart gloo.Error = "subprocess: start" ErrSubprocessReadStdout gloo.Error = "subprocess: read stdout" ErrSubprocess gloo.Error = "subprocess" )
Errors emitted by Subprocess. Each wraps the underlying cause and the process name, so callers can match with errors.Is(err, ErrSubprocessStart) etc.
Variables ¶
This section is empty.
Functions ¶
func Accumulate ¶
Accumulate creates a Command that collects all input, processes it, then emits output. The function receives the complete input as a slice and returns the complete output as a slice.
This is the pattern for commands that must see all input before producing any output. The input/output types are the same — items go in, (possibly reordered/trimmed) items come out.
Shell equivalents: sort, tac, tail -n, shuf
Example — building a Sort command:
func Sort() gloo.Command[string, string] {
return patterns.Accumulate(func(lines []string) ([]string, error) {
sort.Strings(lines)
return lines, nil
})
}
Example — building a Tail command:
func Tail(n int) gloo.Command[string, string] {
return patterns.Accumulate(func(lines []string) ([]string, error) {
if len(lines) <= n {
return lines, nil
}
return lines[len(lines)-n:], nil
})
}
Example ¶
ExampleAccumulate builds a shell-`sort`-like command: all input is collected, processed as one slice, then re-emitted.
package main
import (
"os"
"sort"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
sortLines := patterns.Accumulate(func(lines []string) ([]string, error) {
sort.Strings(lines)
return lines, nil
})
src := gloo.SliceSource([]string{"cherry", "apple", "banana"})
_, _ = gloo.Run(src, gloo.WriteTo(os.Stdout), sortLines)
}
Output: apple banana cherry
func Aggregate ¶
Aggregate creates a Command that reduces all input items to a single output value. The function receives the complete input as a slice and returns one result. The output stream contains exactly one item.
This is the pattern for commands that consume everything and produce a summary. Unlike Accumulate, the output type can differ from the input type.
Shell equivalents: wc -l, wc -w, sha256sum, md5sum, cksum
Example — building a Wc command:
func Wc() gloo.Command[string, int] {
return patterns.Aggregate(func(lines []string) (int, error) {
return len(lines), nil
})
}
Example — building a Sha256sum command:
func Sha256sum() gloo.Command[[]byte, string] {
return patterns.Aggregate(func(chunks [][]byte) (string, error) {
h := sha256.New()
for _, chunk := range chunks {
h.Write(chunk)
}
return hex.EncodeToString(h.Sum(nil)), nil
})
}
Example ¶
ExampleAggregate builds a shell-`wc -l`-like command: all input reduces to a single value, and the output type can differ from the input type.
package main
import (
"fmt"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
countLines := patterns.Aggregate(func(lines []string) (int, error) {
return len(lines), nil
})
src := gloo.SliceSource([]string{"one", "two", "three"})
result, _ := gloo.Chain(src).To(countLines).Collect()
fmt.Println(result.([]int)[0])
}
Output: 3
func Drop ¶
Drop discards the first n items and passes the rest through — the shell `tail -n +N` (skip a header, drop a prefix). Unlike Head it has no early exit: it must read every item, so it does not stop the upstream.
gloo.Compose(patterns.Drop[string](1)).To(sortLines) // skip a CSV header
func Expand ¶
Expand creates a Command that maps each input item to zero or more output items. The function receives one item and returns a slice of results. Output order follows input order — all outputs from item 1 appear before item 2's outputs.
This is the pattern for commands that split, unfold, or fan out each input line.
Shell equivalents: fold (wrap long lines), split-like operations, xargs -n1
Example — building a Split command (split on delimiter):
func Split(delim string) gloo.Command[string, string] {
return patterns.Expand(func(line string) ([]string, error) {
return strings.Split(line, delim), nil
})
}
Example — building a Words command (emit each word as a separate line):
func Words() gloo.Command[string, string] {
return patterns.Expand(func(line string) ([]string, error) {
return strings.Fields(line), nil
})
}
Example ¶
ExampleExpand builds a word-splitting command (like `xargs -n1`): one input line fans out to zero or more output lines.
package main
import (
"os"
"strings"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
words := patterns.Expand(func(line string) ([]string, error) {
return strings.Fields(line), nil
})
src := gloo.SliceSource([]string{"hello world", "foo bar baz"})
_, _ = gloo.Run(src, gloo.WriteTo(os.Stdout), words)
}
Output: hello world foo bar baz
func Fail ¶ added in v0.1.6
Fail creates a Command that emits exactly one error and nothing else, stopping its upstream. It is the canonical way to surface a construction failure (an invalid pattern, an unparsable flag) from a command factory that cannot return an error itself: the pipeline still composes, and the failure arrives where every other failure does — on the stream.
Shell equivalent: a tool that prints a diagnostic and exits non-zero before reading input.
Example — a grep whose pattern does not compile:
func Grep(pattern string) gloo.Command[string, string] {
re, err := regexp.Compile(pattern)
if err != nil {
return patterns.Fail[string](err)
}
return patterns.Filter(func(line string) (bool, error) { return re.MatchString(line), nil })
}
func Filter ¶
Filter creates a Command that keeps or drops items based on a predicate. The function is stateless — it decides per item with no memory of previous items.
This is the pattern for commands where each input line either passes through unchanged or is dropped entirely.
Shell equivalents: grep, grep -v
Example — building a Grep command:
func Grep(pattern string) gloo.Command[string, string] {
return patterns.Filter(func(line string) (bool, error) {
return strings.Contains(line, pattern), nil
})
}
Example ¶
ExampleFilter builds a shell-`grep`-like command: each line either passes through unchanged or is dropped.
package main
import (
"os"
"strings"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
grep := patterns.Filter(func(line string) (bool, error) {
return strings.Contains(line, "red"), nil
})
src := gloo.SliceSource([]string{"apple red", "banana yellow", "cherry red"})
_, _ = gloo.Run(src, gloo.WriteTo(os.Stdout), grep)
}
Output: apple red cherry red
func FilterMap ¶ added in v0.1.6
FilterMap creates a Command that transforms each item AND decides whether to emit it — one pass, one function. The function returns the transformed value, whether to keep it, and any error.
This is the pattern for commands that must transform and select together, where composing Filter after Map would lose the information the decision needs (or compute the transform twice).
Shell equivalents: grep -n (emit "N:line" only for matches), sed -n s//p
Example — building a numbering Grep:
func GrepN(pattern string) gloo.Command[string, string] {
return patterns.StatefulFilterMap(func() func(string) (string, bool, error) {
n := 0
return func(line string) (string, bool, error) {
n++
if !strings.Contains(line, pattern) {
return "", false, nil
}
return fmt.Sprintf("%d:%s", n, line), true, nil
}
})
}
func Head ¶
Head emits the first n items and stops the upstream — the shell `head -n`. It is a named alias for Take, provided because pipelines read more clearly with the familiar command name:
gloo.Compose(grep("error")).To(patterns.Head[string](10))
Like Take, Head over an infinite or huge source terminates promptly and reads only a small prefix. Use Take directly when the early-exit count is not a "head" in spirit (e.g. grep -m, sed q).
func Map ¶
Map creates a Command that transforms each input item independently. The function is stateless — it receives one item and returns one item.
This is the pattern for commands where each input line produces exactly one output line with no memory of previous lines.
Shell equivalents: tr, sed s/old/new/, cut -d -f, basename, dirname
Example — building a Tr command:
func Tr(from, to string) gloo.Command[string, string] {
return patterns.Map(func(line string) (string, error) {
return translate(line, from, to), nil
})
}
Example ¶
ExampleMap builds a shell-`tr`-like command: each input line is transformed independently, one line in, one line out.
package main
import (
"os"
"strings"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
upper := patterns.Map(func(line string) (string, error) {
return strings.ToUpper(line), nil
})
src := gloo.SliceSource([]string{"hello", "world"})
_, _ = gloo.Run(src, gloo.WriteTo(os.Stdout), upper)
}
Output: HELLO WORLD
func MergeSorted ¶ added in v0.1.6
MergeSorted merges two already-sorted streams into one sorted stream — the two-input primitive behind `sort -m` and the streaming half of comm/join. Both inputs are owned by the merge: a downstream stop or an error on either input stops and drains BOTH, so one Discard still collapses the whole graph.
On the first error from either input the error is forwarded and the merge stops — a broken input makes the merged order meaningless.
func Run ¶ added in v0.1.0
Run executes a Source[[]byte] or Command[[]byte, []byte] and writes each output item as a line to os.Stdout, returning any execution error.
Intended for example tests where the // Output: matcher reads stdout — an example asserts success by handling the returned error (test files may panic). For commands that need stream input, Run supplies an empty input stream — useful for self-sourcing commands (e.g. those that read files configured via opts).
Example ¶
ExampleRun executes a Source and writes each item as a line to stdout — the // Output: matcher observes Run's os.Stdout path end-to-end.
src := gloo.SliceSource([][]byte{[]byte("hello"), []byte("world")})
if err := Run(src); err != nil {
panic(err)
}
Output: hello world
func Scan ¶ added in v0.1.6
func Scan[In, Out any]( factory func() (step func(In) ([]Out, error), end func() ([]Out, error)), ) gloo.Command[In, Out]
Scan creates a Command with per-Execute state that may emit zero or more outputs per input item, plus a final batch when the input ends. The factory returns the pair (step, end): step handles one item; end flushes whatever state remains once the stream closes.
This is the pattern for commands whose output depends on stream BOUNDARIES — state that must be emitted when the input ends, which Map/Filter cannot express and Accumulate can only express by buffering everything.
Shell equivalents: uniq -c (flush the last group), awk END blocks, grep -A/-B context windows, paste -s
Example — building uniq -c:
func UniqC() gloo.Command[string, string] {
return patterns.Scan(func() (func(string) ([]string, error), func() ([]string, error)) {
var cur string
n := 0
step := func(line string) ([]string, error) {
if n > 0 && line == cur {
n++
return nil, nil
}
out := group(n, cur) // previous group, if any
cur, n = line, 1
return out, nil
}
end := func() ([]string, error) { return group(n, cur), nil }
return step, end
})
}
func StatefulFilter ¶
StatefulFilter creates a Command where each Execute call gets fresh state. The factory is called once per Execute, returning a predicate function with its own captured state. This ensures the command is reusable as a value.
This is the pattern for commands that filter but need to track something across items (a counter, previous line, etc.).
Shell equivalents: head -n (counter), uniq (previous line)
Example — building a Head command:
func Head(n int) gloo.Command[string, string] {
return patterns.StatefulFilter(func() func(string) (bool, error) {
count := 0
return func(line string) (bool, error) {
count++
return count <= n, nil
}
})
}
Example — building a Uniq command:
func Uniq() gloo.Command[string, string] {
return patterns.StatefulFilter(func() func(string) (bool, error) {
var prev string
first := true
return func(line string) (bool, error) {
if first || line != prev {
prev = line
first = false
return true, nil
}
return false, nil
}
})
}
Example ¶
ExampleStatefulFilter builds a shell-`head -n 2`-like command: the factory runs once per Execute, so the line counter starts fresh for each pipeline.
package main
import (
"os"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
head := patterns.StatefulFilter(func() func(string) (bool, error) {
count := 0
return func(string) (bool, error) {
count++
return count <= 2, nil
}
})
src := gloo.SliceSource([]string{"one", "two", "three", "four"})
_, _ = gloo.Run(src, gloo.WriteTo(os.Stdout), head)
}
Output: one two
func StatefulFilterMap ¶ added in v0.1.6
func StatefulFilterMap[In, Out any](factory func() func(In) (Out, bool, error)) gloo.Command[In, Out]
StatefulFilterMap is FilterMap where each Execute call gets fresh state from the factory, keeping the command reusable as a value.
func StatefulMap ¶
StatefulMap creates a Command where each Execute call gets fresh state. The factory is called once per Execute, returning a transform function with its own captured state. This ensures the command is reusable as a value.
This is the pattern for commands that transform each item but need to track something across items (a counter, running total, etc.).
Shell equivalents: nl (line counter), paste (position tracking)
Example — building an Nl command:
func Nl() gloo.Command[string, string] {
return patterns.StatefulMap(func() func(string) (string, error) {
n := 0
return func(line string) (string, error) {
n++
return fmt.Sprintf("%d\t%s", n, line), nil
}
})
}
Example ¶
ExampleStatefulMap builds a shell-`nl`-like command: the factory runs once per Execute, giving each pipeline its own fresh counter.
package main
import (
"fmt"
"os"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
numbered := patterns.StatefulMap(func() func(string) (string, error) {
n := 0
return func(line string) (string, error) {
n++
return fmt.Sprintf("%d %s", n, line), nil
}
})
src := gloo.SliceSource([]string{"alpha", "beta"})
_, _ = gloo.Run(src, gloo.WriteTo(os.Stdout), numbered)
}
Output: 1 alpha 2 beta
func Subprocess ¶
func Subprocess(name ProcessName, args ...ProcessArg) gloo.Command[[]byte, []byte]
Subprocess creates a Command that forks an external process, streaming the input stream to the child's stdin while capturing the child's stdout as the output stream. Both directions run concurrently, exactly like a process in a shell pipeline: the child starts immediately and produces output while input is still arriving, so memory stays bounded regardless of input size.
Stderr passes through to the parent's stderr. Non-zero exit codes propagate as errors. A downstream stop (the SIGPIPE analogue) or context cancellation closes the pipes and sends SIGTERM, escalating to SIGKILL after a grace period — so `subprocess | Take(3)` terminates the child promptly.
If the child stops reading stdin early (e.g. head -1 exiting), the upstream is STOPPED and its remaining output discarded — so even `infinite | child` terminates, exactly as `yes | head -1` does — and the pipeline result is determined by the child's exit code, mirroring shell SIGPIPE semantics.
This pattern is reserved for commands that wrap irreplaceable external tools (e.g., perl, git). Per constitution INV-1, subprocess usage is discouraged — if the algorithm can be implemented in pure Go, it must be.
Example — building a Perl command:
func Perl(script string) gloo.Command[[]byte, []byte] {
return patterns.Subprocess("perl", "-e", patterns.ProcessArg(script))
}
Example ¶
ExampleSubprocess wraps an external tool as a pipeline command: input lines stream to the child's stdin, its stdout becomes the output stream. Reserved for irreplaceable external tools — prefer pure-Go patterns.
upper := Subprocess("tr", "a-z", "A-Z")
src := gloo.SliceSource([][]byte{[]byte("hello"), []byte("world")})
_, _ = gloo.Run(src, gloo.ByteWriteTo(os.Stdout), upper)
Output: HELLO WORLD
func Take ¶
Take emits the first n items of its input and then stops the upstream — the framework equivalent of `head -n` exiting and killing its feeders via SIGPIPE.
This is the early-termination primitive that StatefulFilter cannot express: a filter may drop items but can never declare "I need nothing more", so a filter-based head over an infinite or huge source reads everything. Take actively tears the upstream down, so `Take(3)` over an infinite source terminates instantly and leaks no goroutines, and `FileSource | Take(3)` reads only a small prefix of the file.
Stages DOWNSTREAM of Take run to completion normally: `Take(3)` composed before a sort still yields three sorted lines, exactly as `seq inf | head -3 | sort` does. Take is the building block for head, grep -m, sed q, and any user-defined early exit.
Take stops at the FIRST upstream error: the error is forwarded and nothing after it is emitted — unlike the per-item transforms (Map, Filter, Expand, Tap), which continue past error items. An early exit that kept reading a broken input would defeat its purpose.
n <= 0 emits nothing and stops the upstream immediately.
Example ¶
ExampleTake builds a shell-`head`-like command that stops its upstream.
package main
import (
"context"
"fmt"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
src := gloo.SliceSource([]int{10, 20, 30, 40, 50})
got, _ := patterns.Take[int](2).Execute(context.Background(), src.Stream(context.Background())).Collect()
fmt.Println(got)
}
Output: [10 20]
func Tap ¶
Tap creates a Command that passes each item through unchanged while calling a side-effect function. An error from the side-effect is emitted as an error item in place of that item; the stream itself CONTINUES (later items are still observed), exactly like the other per-item patterns — it is the consuming terminal that stops a pipeline at its first error.
This is the pattern for commands that observe the stream without modifying it — writing to files, logging, metrics collection.
Shell equivalents: tee (write to file while passing through)
Example — building a Tee command (open once, append each line):
func Tee(w io.Writer) gloo.Command[[]byte, []byte] {
return patterns.Tap(func(line []byte) error {
return gloo.Output(w, line)
})
}
Example ¶
ExampleTap builds a shell-`tee`-like observer: items pass through unchanged while a side effect sees each one.
package main
import (
"fmt"
"os"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
logged := patterns.Tap(func(line string) error {
fmt.Println("seen:", line)
return nil
})
// The sink buffers its writes, so the tap's log lines appear first.
src := gloo.SliceSource([]string{"alpha", "beta"})
_, _ = gloo.Run(src, gloo.WriteTo(os.Stdout), logged)
}
Output: seen: alpha seen: beta alpha beta
Types ¶
type Count ¶ added in v0.1.0
type Count int
Count is the number of items a prefix pattern (Take, Head, Drop) emits or skips — the N of `head -n N`.
type ProcessArg ¶ added in v0.1.0
type ProcessArg string
ProcessArg is a single command-line argument passed to the child process.
type ProcessName ¶ added in v0.1.0
type ProcessName string
ProcessName names the executable a Subprocess runs — the argv[0] handed to the OS (resolved via PATH exactly as exec.Command does).
Source Files