pipeline

Documentation

Overview

Package pipeline provides a framework for creating pipelines.

Index

• type IPipeline
  • func New[In any, Out any](name string, description string, stages []Stage[In, Out]) IPipeline[In, Out]
• type Number
• type Pipeline
  • func (p *Pipeline[In, Out]) GetDescription() string
  • func (p *Pipeline[In, Out]) GetName() string
  • func (p *Pipeline[In, Out]) GetState() status.Status
  • func (p *Pipeline[In, Out]) Run(ctx context.Context, in []In) (out []Out, err error)
  • func (p *Pipeline[In, Out]) SetState(state status.Status)
• type Stage

Types

type IPipeline

type IPipeline[In any, Out any] interface {
	shared.IMeta[In]

	Run(ctx context.Context, in []In) (out []Out, err error)
}

IPipeline defines what a `Pipeline` must do.

func New

func New[In any, Out any](
	name string,
	description string,
	stages []Stage[In, Out],
) IPipeline[In, Out]

New returns a new pipeline.

type Number

type Number struct {
	// Numbers to be processed.
	Numbers []int `json:"numbers"`
}

Number is a simple struct to be used in the tests.

type Pipeline

type Pipeline[In any, Out any] struct {
	// Description of the processor.
	Description string `json:"description"`

	// Name of the processor.
	Name string `json:"name"`

	// Adapters to be used in the pipeline.
	Adapters map[string]adapter.IAdapter[In] `json:"adapters"`

	// Control the pipeline.
	Control chan string `json:"-"`

	// Progress of the pipeline.
	Progress int `json:"progress"`

	// Stages to be used in the pipeline.
	Stages []Stage[In, Out] `json:"stages"`

	// State of the pipeline.
	State status.Status `json:"state"`
}

Pipeline definition.

func (*Pipeline[In, Out]) GetDescription

func (p *Pipeline[In, Out]) GetDescription() string

GetDescription returns the `Description` of the processor.

func (*Pipeline[In, Out]) GetName

func (p *Pipeline[In, Out]) GetName() string

GetName returns the `Name` of the processor.

func (*Pipeline[In, Out]) GetState

func (p *Pipeline[In, Out]) GetState() status.Status

GetState returns the `State` of the processor.

func (*Pipeline[In, Out]) Run

func (p *Pipeline[In, Out]) Run(ctx context.Context, in []In) (out []Out, err error)

Run the pipeline.

func (*Pipeline[In, Out]) SetState

func (p *Pipeline[In, Out]) SetState(state status.Status)

SetState sets the `State` of the processor.

type Stage

type Stage[In any, Out any] struct {
	// Concurrent determines whether the stage should be run concurrently.
	Concurrent bool `json:"concurrent"`

	// Processors to be run in the stage.
	Processors []processor.IProcessor[In, Out] `json:"processors"`
}

Stage definition.