pipeline

Documentation

Index

• func InputsFromFiles(files []string) func() (PipelineEntryInfo, io.ReadCloser, error)
• func ListPipelineSteps() []func() Step
• func RegisterPipelineStep(op string, step func() Step)
• func Run(ctx *PipelineContext) error
• type DefaultPipelineEntryInfo
  • func (pinfo DefaultPipelineEntryInfo) GetName() string
• type ForkStep
  • func (fork ForkStep) Flush(ctx *PipelineContext) error
  • func (ForkStep) Help()
  • func (ForkStep) Name() string
  • func (fork ForkStep) Run(ctx *PipelineContext) error
• type Pipeline
  • func (p *Pipeline) UnmarshalJSON(in []byte) error
  • func (p *Pipeline) UnmarshalYAML(in []byte) error
• type PipelineContext
  • func NewContext(lsctx *ls.Context, env map[string]string, pipeline Pipeline, ...) *PipelineContext
  • func (ctx *PipelineContext) FlushNext() error
  • func (ctx *PipelineContext) HasNext() bool
  • func (ctx *PipelineContext) Next() error
  • func (ctx *PipelineContext) SetGraph(g *lpg.Graph) *PipelineContext
• type PipelineEntryInfo
• type PipelineError
  • func (e PipelineError) Error() string
  • func (e PipelineError) Unwrap() error
• type ReadGraphStep
  • func (ReadGraphStep) Flush(*PipelineContext) error
  • func (ReadGraphStep) Help()
  • func (ReadGraphStep) Name() string
  • func (rd ReadGraphStep) Run(pipeline *PipelineContext) error
• type Step
  • func ReadPipeline(file string) ([]Step, error)
  • func UnmarshalPipeline(stepMarshals []stepMarshal) ([]Step, error)
• type StepFunc
  • func (f StepFunc) Run(ctx *PipelineContext) error
• type UniqueStep
  • func (UniqueStep) Flush(*PipelineContext) error
  • func (UniqueStep) Help()
  • func (UniqueStep) Name() string
  • func (rd UniqueStep) Run(pipeline *PipelineContext) error

Functions

func InputsFromFiles

func InputsFromFiles(files []string) func() (PipelineEntryInfo, io.ReadCloser, error)

InputsFromFiles returns a function that will read input files sequentially for pipeline Run inputs func

func ListPipelineSteps

func ListPipelineSteps() []func() Step

func RegisterPipelineStep

func RegisterPipelineStep(op string, step func() Step)

func Run

func Run(ctx *PipelineContext) error

Run pipeline

Types

type DefaultPipelineEntryInfo

type DefaultPipelineEntryInfo struct {
	Name string
}

func (DefaultPipelineEntryInfo) GetName

func (pinfo DefaultPipelineEntryInfo) GetName() string

type ForkStep

type ForkStep struct {
	Steps map[string]Pipeline `json:"pipelines" yaml:"pipelines"`
}

func (ForkStep) Flush

func (fork ForkStep) Flush(ctx *PipelineContext) error

func (ForkStep) Help

func (ForkStep) Help()

func (ForkStep) Name

func (ForkStep) Name() string

func (ForkStep) Run

func (fork ForkStep) Run(ctx *PipelineContext) error

type Pipeline

type Pipeline []Step

func (*Pipeline) UnmarshalJSON

func (p *Pipeline) UnmarshalJSON(in []byte) error

func (*Pipeline) UnmarshalYAML

func (p *Pipeline) UnmarshalYAML(in []byte) error

type PipelineContext

type PipelineContext struct {
	*ls.Context
	Graph       *lpg.Graph
	Roots       []*lpg.Node
	NextInput   func() (PipelineEntryInfo, io.ReadCloser, error)
	CurrentStep int
	Steps       []Step
	Properties  map[string]interface{}
	Env         map[string]string
	GraphOwner  *PipelineContext
	ErrorLogger func(*PipelineContext, error)
	EntryLogger func(*PipelineContext, map[string]any)
	InfoLogger  func(*PipelineContext, map[string]any)
	// If any goroutines are started with this pipeline, waitgroup is used to wait for them
	Wait sync.WaitGroup
	Err  chan error
}

func NewContext

func NewContext(lsctx *ls.Context, env map[string]string, pipeline Pipeline, initialGraph *lpg.Graph, inputs func() (PipelineEntryInfo, io.ReadCloser, error)) *PipelineContext

create new pipeline context with an optional initial graph and inputs func

func (*PipelineContext) FlushNext

func (ctx *PipelineContext) FlushNext() error

func (*PipelineContext) HasNext

func (ctx *PipelineContext) HasNext() bool

HasNext returns if there is a next step in the pipeline

func (*PipelineContext) Next

func (ctx *PipelineContext) Next() error

func (*PipelineContext) SetGraph

func (ctx *PipelineContext) SetGraph(g *lpg.Graph) *PipelineContext

type PipelineEntryInfo

type PipelineEntryInfo interface {
	GetName() string
}

type PipelineError

type PipelineError struct {
	Wrapped error
	Step    int
}

func (PipelineError) Error

func (e PipelineError) Error() string

func (PipelineError) Unwrap

func (e PipelineError) Unwrap() error

type ReadGraphStep

type ReadGraphStep struct {
	Format string `json:"format" yaml:"format"`
}

func (ReadGraphStep) Flush

func (ReadGraphStep) Flush(*PipelineContext) error

func (ReadGraphStep) Help

func (ReadGraphStep) Help()

func (ReadGraphStep) Name

func (ReadGraphStep) Name() string

func (ReadGraphStep) Run

func (rd ReadGraphStep) Run(pipeline *PipelineContext) error

type Step

type Step interface {
	Name() string
	Run(*PipelineContext) error
	Flush(*PipelineContext) error
}

func ReadPipeline

func ReadPipeline(file string) ([]Step, error)

func UnmarshalPipeline

func UnmarshalPipeline(stepMarshals []stepMarshal) ([]Step, error)

type StepFunc

type StepFunc func(*PipelineContext) error

func (StepFunc) Run

func (f StepFunc) Run(ctx *PipelineContext) error

type UniqueStep

type UniqueStep struct {
	Labels []string `json:"labels" yaml:"labels"`
}

func (UniqueStep) Flush

func (UniqueStep) Flush(*PipelineContext) error

func (UniqueStep) Help

func (UniqueStep) Help()

func (UniqueStep) Name

func (UniqueStep) Name() string

func (UniqueStep) Run

func (rd UniqueStep) Run(pipeline *PipelineContext) error