pipeline

Documentation

Index

• Variables
• func Link[T any, A node.OutputNode[T], B node.InputNode[T]](nodeA A, portA string, nodeB B, portB string) error
• func LinkAny(nodeA node.Node, portA string, nodeB node.Node, portB string) error
• func LinkWithBufferSize[T any, A node.OutputNode[T], B node.InputNode[T]](nodeA A, portA string, nodeB B, portB string, bufferSize int) error
• type BuildOption
  • func WithObservation(mode ObservationMode) BuildOption
• type Builder
  • func NewBuilder() *Builder
  • func (b *Builder) Build(geo *Geometry, options ...BuildOption) (*Pipeline, error)
• type ChanEdge
  • func NewChanEdge[T any](bufferSize int) *ChanEdge[T]
  • func (e *ChanEdge[T]) Close()
  • func (e *ChanEdge[T]) Pull(ctx context.Context) (T, error)
  • func (e *ChanEdge[T]) Push(ctx context.Context, item T) error
• type Description
  • func (d Description) Clone() Description
• type EdgeDef
• type EdgeDescription
• type EdgeSnapshot
• type Geometry
  • func NewGeometry() *Geometry
  • func (g *Geometry) AddEdge(fromNode, fromPort, toNode, toPort string) error
  • func (g *Geometry) AddEdgeDef(definition EdgeDef) error
  • func (g *Geometry) AddNode(id string, n node.Node) error
  • func (g *Geometry) AddNodeDef(definition NodeDef) error
  • func (g *Geometry) AddResourceCloser(closer func() error) error
  • func (g *Geometry) Close() error
  • func (g *Geometry) Description() Description
  • func (g *Geometry) Edges() []EdgeDef
  • func (g *Geometry) Nodes() []NodeDef
  • func (g *Geometry) SetNodeDescription(id string, description NodeDescription) error
• type NodeDef
• type NodeDescription
• type NodeSnapshot
• type ObservationMode
• type Pipeline
  • func New(nodes ...node.Node) (*Pipeline, error)
  • func (p *Pipeline) Close() error
  • func (p *Pipeline) Description() Description
  • func (p *Pipeline) Prepare(ctx context.Context) error
  • func (p *Pipeline) Run(ctx context.Context) error
  • func (p *Pipeline) Snapshot() Snapshot
• type Snapshot

Variables

var ErrInvalidPipeline = errors.New("invalid pipeline")

Functions

func Link

func Link[T any, A node.OutputNode[T], B node.InputNode[T]](nodeA A, portA string, nodeB B, portB string) error

func LinkAny

func LinkAny(nodeA node.Node, portA string, nodeB node.Node, portB string) error

func LinkWithBufferSize

func LinkWithBufferSize[T any, A node.OutputNode[T], B node.InputNode[T]](nodeA A, portA string, nodeB B, portB string, bufferSize int) error

LinkWithBufferSize connects two nodes using a channel edge with the requested capacity. A small capacity is useful for pipelines that must apply strict backpressure to large packets or frames.

Types

type BuildOption

type BuildOption func(*buildConfig)

func WithObservation

func WithObservation(mode ObservationMode) BuildOption

type Builder

type Builder struct{}

func NewBuilder

func NewBuilder() *Builder

func (*Builder) Build

func (b *Builder) Build(geo *Geometry, options ...BuildOption) (*Pipeline, error)

type ChanEdge

type ChanEdge[T any] struct {
	// contains filtered or unexported fields
}

func NewChanEdge

func NewChanEdge[T any](bufferSize int) *ChanEdge[T]

func (*ChanEdge[T]) Close

func (e *ChanEdge[T]) Close()

func (*ChanEdge[T]) Pull

func (e *ChanEdge[T]) Pull(ctx context.Context) (T, error)

func (*ChanEdge[T]) Push

func (e *ChanEdge[T]) Push(ctx context.Context, item T) error

type Description

type Description struct {
	Nodes []NodeDescription
	Edges []EdgeDescription
}

func (Description) Clone

func (d Description) Clone() Description

Clone returns an independent copy that shares no state with d.

type EdgeDef

type EdgeDef = EdgeDescription

type EdgeDescription

type EdgeDescription struct {
	FromNode       string
	FromPort       string
	ToNode         string
	ToPort         string
	Stream         media.StreamInfo
	ProgressSource bool
}

type EdgeSnapshot

type EdgeSnapshot struct {
	Description EdgeDescription
	Items       uint64
	Bytes       uint64
	Samples     uint64
	MediaTime   time.Duration
}

type Geometry

type Geometry struct {
	// contains filtered or unexported fields
}

Geometry owns all added nodes until Builder.Build transfers them to a Pipeline. Close releases nodes when a negotiated geometry is abandoned.

func NewGeometry

func NewGeometry() *Geometry

func (*Geometry) AddEdge

func (g *Geometry) AddEdge(fromNode, fromPort, toNode, toPort string) error

func (*Geometry) AddEdgeDef

func (g *Geometry) AddEdgeDef(definition EdgeDef) error

func (*Geometry) AddNode

func (g *Geometry) AddNode(id string, n node.Node) error

func (*Geometry) AddNodeDef

func (g *Geometry) AddNodeDef(definition NodeDef) error

func (*Geometry) AddResourceCloser

func (g *Geometry) AddResourceCloser(closer func() error) error

AddResourceCloser registers a shared resource (such as a worker pool held by several nodes) to be closed once every node in the eventual Pipeline has closed, or immediately if the geometry is abandoned instead of built.

func (*Geometry) Close

func (g *Geometry) Close() error

func (*Geometry) Description

func (g *Geometry) Description() Description

func (*Geometry) Edges

func (g *Geometry) Edges() []EdgeDef

func (*Geometry) Nodes

func (g *Geometry) Nodes() []NodeDef

func (*Geometry) SetNodeDescription

func (g *Geometry) SetNodeDescription(id string, description NodeDescription) error

type NodeDef

type NodeDef struct {
	ID          string
	Node        node.Node
	Description NodeDescription
}

type NodeDescription

type NodeDescription struct {
	ID     string
	Role   manifest.NodeType
	Plugin string
	// Configuration is excluded from JSON: plugin config types are free to
	// hold non-marshalable values (e.g. codec-flac's Apodizations, which are
	// func([]float64)), so this live object was never meant as a wire
	// format. Use cliflag.DescribeStruct for a safe, field-by-field view.
	Configuration registry.Configuration `json:"-"`
	Resources     registry.ResourceGrant
	Inputs        []media.StreamInfo
	Outputs       []media.StreamInfo
	AutoInserted  bool
}

type NodeSnapshot

type NodeSnapshot struct {
	Description NodeDescription
	State       string
	StartedAt   time.Time
	FinishedAt  time.Time
	Elapsed     time.Duration
	Error       string
}

type ObservationMode

type ObservationMode uint8

const (
	ObservationOff ObservationMode = iota
	ObservationProgress
	ObservationMetrics
)

type Pipeline

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline owns its nodes for their complete lifecycle. A Pipeline is single-use: Run may be called exactly once, and always closes every node before returning.

func New

func New(nodes ...node.Node) (*Pipeline, error)

func (*Pipeline) Close

func (p *Pipeline) Close() error

func (*Pipeline) Description

func (p *Pipeline) Description() Description

func (*Pipeline) Prepare

func (p *Pipeline) Prepare(ctx context.Context) error

Prepare performs resource-dependent node setup. It is safe to call more than once; Run calls it automatically when needed.

func (*Pipeline) Run

func (p *Pipeline) Run(ctx context.Context) error

func (*Pipeline) Snapshot

func (p *Pipeline) Snapshot() Snapshot

type Snapshot

type Snapshot struct {
	State      string
	StartedAt  time.Time
	FinishedAt time.Time
	Elapsed    time.Duration
	Nodes      []NodeSnapshot
	Edges      []EdgeSnapshot
}