gloo

func Pipe ¶ added in v0.0.4

func Pipe[A, B, C any](cmd1 Command[A, B], cmd2 Command[B, C]) Command[A, C]

Pipe composes two commands into a new command. It connects the output of cmd1 to the input of cmd2. The result is immutable and safe to reuse across pipelines.

package main

import (
	"context"
	"fmt"
	"strings"

	gloo "github.com/gloo-foo/framework"
	"github.com/gloo-foo/framework/patterns"
)

func main() {
	ctx := context.Background()
	src := gloo.SliceSource([]string{"apple red", "banana yellow", "cherry red"})

	grep := patterns.Filter(func(line string) (bool, error) {
		return strings.Contains(line, "red"), nil
	})

	upper := patterns.Map(func(line string) (string, error) {
		return strings.ToUpper(line), nil
	})

	composed := gloo.Pipe(grep, upper)
	stream := gloo.From(ctx, src, composed)
	results, _ := stream.Collect()
	for _, r := range results {
		fmt.Println(r)
	}
}

Output:
APPLE RED
CHERRY RED

func Chain ¶ added in v0.0.7

func Chain(source any) FluentPipeline

Chain starts a fluent pipeline from a source using context.Background(). Use .To() to chain commands and .Sink()/.Collect()/.ForEach() to consume.

gloo.Chain(source).To(grepCmd).To(sortCmd).Sink(sink)

For compile-time-checked composition (no reflection, no runtime type errors), prefer Pipe and Compose; Chain trades that safety for a fluent, type-changing syntax until Go ships generic methods.

package main

import (
	"fmt"
	"strings"

	gloo "github.com/gloo-foo/framework"
	"github.com/gloo-foo/framework/patterns"
)

func main() {
	src := gloo.SliceSource([]string{"apple red", "banana yellow", "cherry red"})

	grep := patterns.Filter(func(line string) (bool, error) {
		return strings.Contains(line, "red"), nil
	})

	upper := patterns.Map(func(line string) (string, error) {
		return strings.ToUpper(line), nil
	})

	result, _ := gloo.Chain(src).
		To(grep).
		To(upper).
		Collect()

	for _, item := range result.([]string) {
		fmt.Println(item)
	}
}

Output:
APPLE RED
CHERRY RED

package main

import (
	"fmt"

	gloo "github.com/gloo-foo/framework"
)

func main() {
	src := gloo.SliceSource([]string{"one", "two", "three"})

	_ = gloo.Chain(src).ForEach(func(line string) error {
		fmt.Println(line)
		return nil
	})
}

Output:
one
two
three

package main

import (
	"fmt"
	"os"
	"strings"

	gloo "github.com/gloo-foo/framework"
	"github.com/gloo-foo/framework/patterns"
)

func main() {
	src := gloo.SliceSource([]string{"hello", "world"})

	upper := patterns.Map(func(line string) (string, error) {
		return strings.ToUpper(line), nil
	})

	count, _ := gloo.Chain(src).To(upper).Sink(gloo.WriteTo(os.Stdout))
	fmt.Printf("%d lines written\n", count)
}

Output:
HELLO
WORLD
2 lines written

func ChainContext ¶ added in v0.0.7

func ChainContext(ctx context.Context, source any) FluentPipeline

ChainContext starts a fluent pipeline from a source with an explicit context. The context is inherited by .To() stages. Use .ToContext() to override the context for individual stages. A nil context is treated as context.Background().

Wiring is LAZY: the source and commands are validated for type compatibility at build time (by inspecting method signatures, without invoking them) but the stream is not created until a terminal operation runs. A chain that is built and never consumed therefore starts no goroutines and leaks nothing. A validation failure is recorded and returned by the terminal, never panicked.

func NewParameters ¶ added in v0.0.7

func NewParameters[P, F any](parameters ...any) Parameters[P, F]

NewParameters classifies a heterogeneous argument list into typed positional values, Switch[F] flags, and an ambiguous bin. Files named by File positionals are opened lazily by Reader/ReadersFrom, not here.

func Compose ¶ added in v0.0.7

func Compose[T any](cmd Command[T, T]) Pipeline[T]

Compose starts a same-type command chain from an initial Command[T, T]. Each .To() returns a NEW Pipeline (immutable). The resulting Pipeline implements Command[T, T] and is safe to reuse across multiple pipelines.

pipeline := gloo.Compose(grepCmd).To(sortCmd).To(uniqCmd)
// pipeline implements Command[T, T]

package main

import (
	"fmt"
	"strings"

	gloo "github.com/gloo-foo/framework"
	"github.com/gloo-foo/framework/patterns"
)

func main() {
	src := gloo.SliceSource([]string{"hello", "world"})

	addBang := patterns.Map(func(line string) (string, error) {
		return line + "!", nil
	})

	upper := patterns.Map(func(line string) (string, error) {
		return strings.ToUpper(line), nil
	})

	composed := gloo.Compose(addBang).To(upper)

	result, _ := gloo.Chain(src).To(composed).Collect()
	for _, item := range result.([]string) {
		fmt.Println(item)
	}
}

Output:
HELLO!
WORLD!

func ByteWriteTo ¶ added in v0.0.7

func ByteWriteTo(w io.Writer) Sink[[]byte, int]

ByteWriteTo creates a Sink that writes each []byte item as a line to the given writer.

func ByteWriteToTerminated ¶ added in v0.1.6

func ByteWriteToTerminated(w io.Writer, t Terminator) Sink[[]byte, int]

ByteWriteToTerminated is ByteWriteTo with an explicit record terminator — pass NUL for -z/-0 style output.

func RawWriteTo ¶ added in v0.1.6

func RawWriteTo(w io.Writer) Sink[[]byte, ByteCount]

RawWriteTo creates a Sink that writes each []byte item VERBATIM — no terminator is appended. It is the binary-safe sink: wired to a chunk source (ChunkReaderSource, ChunkFileSource) it reproduces input byte-for-byte, the way `cat` treats a binary file. The result is the total byte count written.

func WriteTo ¶ added in v0.0.7

func WriteTo(w io.Writer) Sink[string, int]

WriteTo creates a Sink that writes each string item as a line to the given writer.

func WriteToTerminated ¶ added in v0.1.6

func WriteToTerminated(w io.Writer, t Terminator) Sink[string, int]

WriteToTerminated is WriteTo with an explicit record terminator — pass NUL for -z/-0 style output.

func ByteFileSource ¶ added in v0.0.7

func ByteFileSource(fs afero.Fs, files []File) Source[[]byte]

ByteFileSource creates a stream of []byte lines from files on the given filesystem. Each line is an independent copy, safe to retain after the next scan.

func ByteFileSourceTerminated ¶ added in v0.1.6

func ByteFileSourceTerminated(fs afero.Fs, files []File, t Terminator) Source[[]byte]

ByteFileSourceTerminated is ByteFileSource splitting records at an explicit terminator — pass NUL for -z/-0 style input.

func ByteReaderSource ¶ added in v0.0.7

func ByteReaderSource(readers []io.Reader) Source[[]byte]

ByteReaderSource creates a stream of []byte lines from readers. Each line is an independent copy, safe to retain after the next scan.

func ByteReaderSourceTerminated ¶ added in v0.1.6

func ByteReaderSourceTerminated(readers []io.Reader, t Terminator) Source[[]byte]

ByteReaderSourceTerminated is ByteReaderSource splitting records at an explicit terminator — pass NUL for -z/-0 style input.

func ChunkFileSource ¶ added in v0.1.6

func ChunkFileSource(fs afero.Fs, files []File, size ChunkSize) Source[[]byte]

ChunkFileSource is ChunkReaderSource over files on the given filesystem.

func ChunkReaderSource ¶ added in v0.1.6

func ChunkReaderSource(readers []io.Reader, size ChunkSize) Source[[]byte]

ChunkReaderSource creates a BINARY-SAFE stream of []byte chunks from readers: bytes pass through verbatim — no line splitting, no terminator stripping — so NUL bytes, partial lines, and missing final newlines survive untouched. Wire it to RawWriteTo for a byte-identical copy (`cat` of a binary file). A size <= 0 uses DefaultChunkSize. Each chunk is independently allocated and safe to retain.

func FileSource ¶ added in v0.0.7

func FileSource(fs afero.Fs, files []File) Source[string]

FileSource creates a stream of string lines from files on the given filesystem.

package main

import (
	"context"
	"fmt"

	"github.com/spf13/afero"

	gloo "github.com/gloo-foo/framework"
)

func main() {
	fs := afero.NewMemMapFs()
	_ = afero.WriteFile(fs, "test.txt", []byte("line one\nline two\nline three"), 0o644)

	ctx := context.Background()
	src := gloo.FileSource(fs, []gloo.File{"test.txt"})
	results, _ := src.Stream(ctx).Collect()
	for _, r := range results {
		fmt.Println(r)
	}
}

Output:
line one
line two
line three

func FileSourceTerminated ¶ added in v0.1.6

func FileSourceTerminated(fs afero.Fs, files []File, t Terminator) Source[string]

FileSourceTerminated is FileSource splitting records at an explicit terminator instead of newlines — pass NUL for -z/-0 style input. Records pass verbatim (no carriage-return stripping).

func ReaderSource ¶ added in v0.0.7

func ReaderSource(readers []io.Reader) Source[string]

ReaderSource creates a stream of string lines from readers.

func ReaderSourceTerminated ¶ added in v0.1.6

func ReaderSourceTerminated(readers []io.Reader, t Terminator) Source[string]

ReaderSourceTerminated is ReaderSource splitting records at an explicit terminator instead of newlines — pass NUL for -z/-0 style input. Records pass verbatim (no carriage-return stripping).

func SliceSource ¶ added in v0.0.7

func SliceSource[T any](items []T) Source[T]

SliceSource creates a Source from an in-memory slice. This is the equivalent of shell here-strings or echo piping.

package main

import (
	"context"
	"fmt"

	gloo "github.com/gloo-foo/framework"
)

func main() {
	ctx := context.Background()
	src := gloo.SliceSource([]string{"alpha", "bravo", "charlie"})

	results, _ := src.Stream(ctx).Collect()
	for _, r := range results {
		fmt.Println(r)
	}
}

Output:
alpha
bravo
charlie

func From ¶ added in v0.0.7

func From[A, B any](ctx context.Context, src Source[A], cmd Command[A, B]) Stream[B]

From wires a source into a command, returning the output stream. The caller owns the returned stream: consume it fully (e.g. via Collect, which drains), or call Discard to abandon it — Discard both stops the upstream and drains, which Stop alone does not (pure rill transform stages must be drained to unblock).

func Generate ¶ added in v0.0.7

func Generate[T any](ctx context.Context, producer producerFunc[T]) Stream[T]

Generate runs a cancellation-aware ORIGIN producer (a Source, with no upstream) and returns its Stream. It is the primitive every origin producer is built on. For a producer derived from an input stream, use GenerateFrom so teardown chains upstream.

Teardown semantics, derived from context.Cause of the producer scope:

• Downstream stop (cause == ErrStopReading): close silently.
• External cancellation (any other cause, e.g. ^C / deadline): emit the cause once — unless the producer already emitted an error — then close.
• Natural completion: close.

The returned Stream's Discard cancels this producer scope with ErrStopReading.

func GenerateFrom ¶ added in v0.0.7

func GenerateFrom[In, Out any](ctx context.Context, upstream Stream[In], producer producerFunc[Out]) Stream[Out]

GenerateFrom is Generate for a producer that derives its output from an upstream stream. A downstream Discard cancels this producer AND tears the upstream down, so one Discard collapses the whole chain (upstream only). Authors pass the INPUT stream rather than a loose stop handle, so the safe teardown is the only thing in reach.

When the producer returns, the upstream is automatically discarded (stopped and drained), so a producer that returns early — after an error, or once it has read all it needs — cannot strand an upstream stage, even one with no cancellation hook of its own (a pure rill transform, a bare Wrap'd channel). An explicit upstream.Discard() inside the producer remains correct and idempotent. The upstream must not be read after the producer returns.

func StreamOf ¶ added in v0.0.7

func StreamOf[T any](items ...T) Stream[T]

StreamOf builds a finished Stream from in-memory values. It is the canonical way to feed synthetic input to a command in a test — the framework analogue of a shell here-string — and backs SliceSource.

func Wrap ¶ added in v0.0.7

func Wrap[T any](ch <-chan rill.Try[T]) Stream[T]

Wrap adapts a standalone channel (an ORIGIN, with no upstream to tear down) into a Stream. StreamOf and a FuncCommand that synthesizes a channel from nothing use it. A transform that wraps a rill stage over an input stream must use WrapFrom so teardown chains upstream.

func WrapFrom ¶ added in v0.0.7

func WrapFrom[In, Out any](ch <-chan rill.Try[Out], upstream Stream[In]) Stream[Out]

WrapFrom adapts the output channel of a pure rill transform into a Stream, chaining teardown to the upstream stream so a downstream Discard propagates. A pure transform has no producer scope of its own — it terminates when its input closes — so forwarding teardown is all that is required. The forward is guarded so it stays idempotent however many times it fires.

out := rill.OrderedMap(in.Chan(), 1, fn)
return gloo.WrapFrom(out, in)