Documentation
¶
Overview ¶
Package pipeline composes and transforms data flowing from sources to consumers. A Pipeline wraps an upstream (a source or another pipeline) and either passes its data through (v1) or applies an operation (future: filter, project, sort, derive, union, compose, join). Pipelines satisfy datasource.Source so any consumer that reads from a source can read from a pipeline without changes — including the TUI's component-binding layer and wrangl's stdout printer.
Hard import rule ¶
This package — and internal/datasource, internal/output, and internal/config — MUST NOT import anything from internal/screen, internal/build, or any tuilib package. The data layer is architecturally independent of the TUI. A CI check enforces this; see .github/workflows/import-boundary.yml.
Violating this rule defeats the point of the wrangl binary: it should compile and run with zero TUI code in its dependency graph.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline is the data layer's named, addressable unit. It wraps a single upstream (source or other pipeline) and either passes Fetch / Subscribe through unchanged (passthrough) or applies a per-item transformation declared by an operator (filter today; project / derive / sort to come).
Pipeline satisfies datasource.Source / StreamingSource by delegation, so the existing binding code in internal/build doesn't need to special-case pipelines vs sources.
Operator hooks:
- transformSnapshot rewrites the value returned by Fetch (one-shot and polled). nil = identity (passthrough).
- transformEvent rewrites or drops a single event from Subscribe. Returns the rewritten event, a `keep` flag (false drops the event silently), and an error to surface as a stream error. nil = identity.
Both hooks default to identity so the zero-value Pipeline is the passthrough we shipped first; operators just override them.
func (*Pipeline) Fetch ¶
Fetch delegates to the upstream's Fetch, then applies the pipeline's snapshot transform (if any). Passthrough pipelines have no transform and return the upstream value unchanged.
func (*Pipeline) Name ¶
Name returns the config-declared name. Used by introspection (--list, --explain).
func (*Pipeline) Refresh ¶
Refresh delegates to the upstream. A passthrough pipeline takes its lifecycle from whatever it wraps.
func (*Pipeline) Subscribe ¶
Subscribe delegates when the upstream is a StreamingSource. Pipelines wrapping non-streaming sources return ErrNotStreaming so the screen falls back to polling, same convention as other composers.
When a transformEvent hook is set, every upstream event is passed through it: kept events forward to consumers, dropped events silently disappear, errors surface as Event{Err}. Without a hook, events forward unchanged (the passthrough case).
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry holds built pipelines indexed by name, alongside the live sources they sit on top of. Callers (screen.New, wrangl main) look up by name and treat sources and pipelines uniformly via the SourceLike interface.
func Build ¶
func Build(prebuilt map[string]ds.Source, sources map[string]*cfg.Source, boundParams map[string]map[string]string) (*Registry, error)
Build constructs every entry in the unified data.sources map. Walks the graph topologically (cycle detection happens at validate-time so the recursive resolve can't loop) and dispatches per-kind via the source's Type string:
- Leaf kinds → ds.BuildLeaf, with merge children pre-resolved.
- Operator kinds → the per-operator constructor in this package (newFilter, newProject, …), with upstreams resolved from the same sources map.
Arguments:
- prebuilt: pre-constructed ds.Source values registered by name before resolution. Production callers pass nil. Test fixtures pass pre-built fakeSource values so operators wired on top of them don't need to go through ds.BuildLeaf.
- sources: the unified sources map. Resolved recursively; operator kinds dispatch through the per-op constructor.
- boundParams: caller-bound parameter values keyed by source name; entries without a key in this map run with all-default / required-blocked params.
The returned Registry exposes one Get(name) lookup that resolves any entry — leaves and operators share the namespace.
func (*Registry) Get ¶
Get resolves a name to its live ds.Source — pipeline first (named pipelines shadow sources of the same name, though the config validator prevents that collision), then source. Returns nil if the name isn't defined.
func (*Registry) Names ¶
Names returns every defined name (sources + pipelines), sorted. Used by --list and friends.
Source Files