gloo

Jul 11, 2026

Changes in this version

+ type ByteCount int64

+ type ChunkSize int

+ const DefaultChunkSize

+ type ExitStatus int

+ func (e ExitStatus) Error() string

+ func (e ExitStatus) Status() int

type Sink

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

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

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

type Source

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

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

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

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

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

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

Jul 3, 2026

Changes in this version

+ type Line string

+ type Terminator byte

Jun 21, 2026

Changes in this version

+ const MaxLineSize

+ const NewLine

+ func Collect[T any](_ context.Context, stream Stream[T]) ([]T, error)

+ func Into[A, B, R any](ctx context.Context, src Source[A], cmd Command[A, B], sink Sink[B, R]) (R, error)

+ func NewLineScanner(r io.Reader) *bufio.Scanner

+ func Output(stdout io.Writer, data []byte) error

+ func OutputString(stdout io.Writer, data string) error

+ func OutputStringWithTerminator(stdout io.Writer, terminator byte, data string) error

+ func OutputWithTerminator(stdout io.Writer, terminator byte, data []byte) error

+ func Pump(ctx context.Context, cmd Command[string, string], stdin io.Reader, ...) (int, error)

+ func PumpBytes(ctx context.Context, cmd Command[[]byte, []byte], stdin io.Reader, ...) (int, error)

+ func RunContext(ctx context.Context, source any, sink any, cmds ...any) (any, error)

type Command

+ Execute func(ctx context.Context, input Stream[In]) Stream[Out]

+ type Error string

+ const ErrFileNotFound

+ const ErrNotCommand

+ const ErrNotForEachFunc

+ const ErrNotSink

+ const ErrNotSource

+ const ErrPipelineConsumed

+ const ErrSinkTypeMismatch

+ const ErrStageTypeMismatch

+ const ErrStopReading

+ func (e Error) Error() string

+ func (e Error) With(err error, args ...any) error

+ type FluentPipeline struct

+ func Chain(source any) *FluentPipeline

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

+ func (p *FluentPipeline) Collect() (any, error)

+ func (p *FluentPipeline) ForEach(fn any) error

+ func (p *FluentPipeline) Sink(sink any) (any, error)

+ func (p *FluentPipeline) SinkContext(ctx context.Context, sink any) (any, error)

+ func (p *FluentPipeline) To(cmd any) *FluentPipeline

+ func (p *FluentPipeline) ToContext(ctx context.Context, cmd any) *FluentPipeline

+ type FuncCommand func(context.Context, Stream[In]) Stream[Out]

+ func (f FuncCommand[In, Out]) Execute(ctx context.Context, input Stream[In]) Stream[Out]

+ type Parameters struct

+ Ambiguous []any

+ Flags F

+ Positional []any

+ Typed []P

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

+ func (p Parameters[P, F]) Reader(stdin io.Reader) (io.ReadCloser, error)

+ func (p Parameters[P, F]) ReaderFrom(fs afero.Fs, stdin io.Reader) (io.ReadCloser, error)

+ func (p Parameters[P, F]) ReadersFrom(fs afero.Fs) ([]io.ReadCloser, error)

type Pipeline

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

+ func (p Pipeline[T]) Execute(ctx context.Context, input Stream[T]) Stream[T]

+ func (p Pipeline[T]) To(cmd Command[T, T]) Pipeline[T]

+ type Sink interface

+ Consume func(ctx context.Context, input Stream[In]) (Res, error)

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

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

+ type Source interface

+ Stream func(ctx context.Context) Stream[Out]

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

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

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

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

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

+ type Stream struct

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

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

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

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

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

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

+ func (s Stream[T]) Chan() <-chan rill.Try[T]

+ func (s Stream[T]) Collect() ([]T, error)

+ func (s Stream[T]) Discard()

Nov 15, 2025

Changes in this version

+ func MustRunChannel[T any](cmd ChannelCommand[T])

+ func ReaderToChannelParsed[T any](ctx context.Context, r io.Reader, out chan<- Row[T]) error

+ func ReaderToChannelWithParser[T any](ctx context.Context, r io.Reader, out chan<- Row[T], ...) error

+ func RunChannelWithContext[T any](ctx context.Context, cmd ChannelCommand[T]) error

+ func RunChannel[T any](cmd ChannelCommand[T]) error

+ type ChannelCommand interface

+ ChannelExecutor func() ChannelExecutor[T]

+ type ChannelExecutor func(ctx context.Context, in <-chan Row[T], out chan<- Row[T]) error

+ func Batch[T any](size int, fn func([]T) ([]T, error)) ChannelExecutor[T]

+ func ParallelMap[T any](workers int, fn func(T) (T, bool, error)) ChannelExecutor[T]

+ func Pipe[T any](first ChannelExecutor[T], rest ...ChannelExecutor[T]) ChannelExecutor[T]

type Command

+ func Pipeline[T any](executor ChannelExecutor[T]) Command

type Inputs

+ func (inputs Inputs[T, O]) ToChannelBytes(ctx context.Context, stdin io.Reader, out chan<- Row[[]byte]) error

+ func (inputs Inputs[T, O]) ToChannelString(ctx context.Context, stdin io.Reader, out chan<- Row[string]) error

+ func (inputs Inputs[T, O]) WrapChannelBytes(executor ChannelExecutor[[]byte]) CommandExecutor

+ func (inputs Inputs[T, O]) WrapChannelString(executor ChannelExecutor[string]) CommandExecutor

+ type Row struct

+ Data T

+ Err error

+ type RowParser interface

+ ParseRow func(line string) error

Nov 7, 2025

Changes in this version

+ func Must(err error)

+ func MustRun(cmd Command)

+ func Run(cmd Command) error

+ func RunWithContext(ctx context.Context, cmd Command) error

+ type AccumulateAndOutputFunc func(lines []string, stdout io.Writer) error

+ type AccumulateAndProcessFunc func(lines []string) []string

+ type Command interface

+ Executor func() CommandExecutor

+ func AccumulateAndOutput(fn AccumulateAndOutputFunc) Command

+ func AccumulateAndProcess(fn AccumulateAndProcessFunc) Command

+ func LineTransform(fn LineTransformFunc) Command

+ func RawCommand(fn CommandExecutor) Command

+ func StatefulLineTransform(fn StatefulLineTransformFunc) Command

+ type CommandExecutor func(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer) error

+ type File string

+ type Inputs struct

+ Ambiguous []any

+ Flags O

+ Positional []T

+ func Initialize[T any, O any](parameters ...any) Inputs[T, O]

+ func (inputs Inputs[T, O]) Close() error

+ func (inputs Inputs[T, O]) Reader(stdin io.Reader) io.Reader

+ func (inputs Inputs[T, O]) Readers() []io.Reader

+ func (inputs Inputs[T, O]) Wrap(executor CommandExecutor) CommandExecutor

+ type LineTransformFunc func(line string) (output string, emit bool)

+ type StatefulLineTransformFunc func(lineNum int64, line string) (output string, emit bool)

+ type Switch interface

+ Configure func(*T)