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 int) gloo.Command[T, T]
- func Expand[In, Out any](fn func(In) ([]Out, error)) gloo.Command[In, Out]
- func Filter[T any](fn func(T) (bool, error)) gloo.Command[T, T]
- func Head[T any](n int) gloo.Command[T, T]
- func Map[In, Out any](fn func(In) (Out, error)) gloo.Command[In, Out]
- func MustRun(target any)
- func StatefulFilter[T any](factory func() func(T) (bool, error)) gloo.Command[T, T]
- func StatefulMap[In, Out any](factory func() func(In) (Out, error)) gloo.Command[In, Out]
- func Subprocess(name string, args ...string) gloo.Command[[]byte, []byte]
- func Take[T any](n int) gloo.Command[T, T]
- func Tap[T any](fn func(T) error) gloo.Command[T, T]
Examples ¶
Constants ¶
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 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 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 MustRun ¶
func MustRun(target any)
MustRun executes a Source[[]byte] or Command[[]byte, []byte] and writes each output item as a line to os.Stdout. It panics on any error.
Intended for example tests where the // Output: matcher reads stdout. For commands that need stream input, MustRun supplies an empty input stream — useful for self-sourcing commands (e.g. those that read files configured via opts).
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 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 ¶
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 remaining input is drained and discarded 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", 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.
package main
import (
"os"
gloo "github.com/gloo-foo/framework"
"github.com/gloo-foo/framework/patterns"
)
func main() {
upper := patterns.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.
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. Errors from the side-effect stop the stream.
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:
func Tee(path string, fs afero.Fs) gloo.Command[[]byte, []byte] {
return patterns.Tap(func(line []byte) error {
return afero.WriteFile(fs, path, line, 0644)
})
}
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 ¶
This section is empty.
Source Files