flow

Documentation

Index

• Variables
• func IsFinishedStatus(sts string) bool
• func RegisterExecutorBuilder(name string, builder func(flow types.Flow) Executor)
• type Controller
  • func NewFlowController(storage Storage) *Controller
  • func (c *Controller) CancelFlow(flowId string) error
  • func (c *Controller) PauseFlow(flowId string) error
  • func (c *Controller) ResumeFlow(flowId string) error
  • func (c *Controller) Shutdown() error
  • func (c *Controller) TriggerFlow(ctx context.Context, flowId string) error
• type DAG
• type Executor
• type MemStorage
  • func (m *MemStorage) DeleteFlow(ctx context.Context, flowId string) error
  • func (m *MemStorage) GetFlow(ctx context.Context, flowId string) (types.Flow, error)
  • func (m *MemStorage) SaveFlow(ctx context.Context, flow types.Flow) error
  • func (m *MemStorage) SaveTask(ctx context.Context, flowId string, task types.Task) error
• type Operator
• type Parameter
• type ResultData
• type Runner
  • func NewRunner(f types.Flow, storage Storage) Runner
• type Storage
  • func NewInMemoryStorage() Storage

Variables

var ErrNotFound = fmt.Errorf("not found")

var (
	ExecutorNotFound = errors.New("executor not found")
)

Functions

func IsFinishedStatus

func IsFinishedStatus(sts string) bool

func RegisterExecutorBuilder

func RegisterExecutorBuilder(name string, builder func(flow types.Flow) Executor)

Types

type Controller

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

func NewFlowController

func NewFlowController(storage Storage) *Controller

func (*Controller) CancelFlow

func (c *Controller) CancelFlow(flowId string) error

func (*Controller) PauseFlow

func (c *Controller) PauseFlow(flowId string) error

func (*Controller) ResumeFlow

func (c *Controller) ResumeFlow(flowId string) error

func (*Controller) Shutdown

func (c *Controller) Shutdown() error

func (*Controller) TriggerFlow

func (c *Controller) TriggerFlow(ctx context.Context, flowId string) error

type DAG

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

type Executor

type Executor interface {
	Setup(ctx context.Context) error
	DoOperation(ctx context.Context, task types.Task) error
	Teardown(ctx context.Context)
}

type MemStorage

type MemStorage struct {
	sync.Map
}

func (*MemStorage) DeleteFlow

func (m *MemStorage) DeleteFlow(ctx context.Context, flowId string) error

func (*MemStorage) GetFlow

func (m *MemStorage) GetFlow(ctx context.Context, flowId string) (types.Flow, error)

func (*MemStorage) SaveFlow

func (m *MemStorage) SaveFlow(ctx context.Context, flow types.Flow) error

func (*MemStorage) SaveTask

func (m *MemStorage) SaveTask(ctx context.Context, flowId string, task types.Task) error

type Operator

type Operator interface {
	Do(ctx context.Context, param *Parameter) error
}

type Parameter

type Parameter struct {
	FlowID  string
	Workdir string
	Result  *ResultData
}

type ResultData

type ResultData struct {
	Result   sync.Map
	TempData sync.Map
}

type Runner

type Runner interface {
	Start(ctx context.Context) error
	Pause() error
	Resume() error
	Cancel() error
}

func NewRunner

func NewRunner(f types.Flow, storage Storage) Runner

type Storage

type Storage interface {
	GetFlow(ctx context.Context, flowId string) (types.Flow, error)
	SaveFlow(ctx context.Context, flow types.Flow) error
	DeleteFlow(ctx context.Context, flowId string) error
	SaveTask(ctx context.Context, flowId string, task types.Task) error
}

func NewInMemoryStorage

func NewInMemoryStorage() Storage