executor

Documentation

Index

• Variables
• type ExecOption
• type FlowExecutor
  • func (exec *FlowExecutor) Create(tplID int, name string, option *ExecOption, nodes []*models.Node, ...) ([]*models.Flow, error)
  • func (exec *FlowExecutor) MergeParams(options []*models.StepOption, overrideParams map[string]interface{})
  • func (exec *FlowExecutor) Pause(flow *models.Flow) error
  • func (exec *FlowExecutor) Run(tplID int, name string, option *ExecOption, nodes []*models.Node, ...) error
  • func (exec *FlowExecutor) RunStep(h handler.Handler, step *models.ActionImpl, stepNum int, ...) ([]*models.NodeState, []*models.NodeState)
  • func (exec *FlowExecutor) SetFlowStatus(flow *models.Flow, status int) error
  • func (exec *FlowExecutor) Start(flow *models.Flow) error
  • func (exec *FlowExecutor) Stop(flow *models.Flow) error
• type Job
• type Worker
  • func NewWorker(key int) *Worker
  • func (w *Worker) Start()
  • func (w *Worker) Stop()
  • func (w *Worker) Submit(f Job) error

Variables

var (
	// Executor is the singleton FlowExecutor instance to run all tasks.
	Executor = &FlowExecutor{}
)

Types

type ExecOption

type ExecOption struct {
	// MaxNum is the max num of nodes in one batch.
	MaxNum     int
	MaxFailNum int
}

ExecOption contains options to execute a task.

type FlowExecutor

type FlowExecutor struct {
}

FlowExecutor dispatches tasks into queues.

func (*FlowExecutor) Create

func (exec *FlowExecutor) Create(tplID int, name string, option *ExecOption,
	nodes []*models.Node, context map[string]interface{}) ([]*models.Flow, error)

Create creates a flow instance with given template id, and nodes.

func (*FlowExecutor) MergeParams

func (exec *FlowExecutor) MergeParams(options []*models.StepOption,
	overrideParams map[string]interface{})

func (*FlowExecutor) Pause

func (exec *FlowExecutor) Pause(flow *models.Flow) error

Pause paused a running flow by setting its status to STOPPED.

func (*FlowExecutor) Run

func (exec *FlowExecutor) Run(tplID int, name string, option *ExecOption,
	nodes []*models.Node, context map[string]interface{}) error

Run creates a flow instance from the given template id, and nodes, and starts the flow instance.

func (*FlowExecutor) RunStep

func (exec *FlowExecutor) RunStep(h handler.Handler, step *models.ActionImpl, stepNum int,
	nstates []*models.NodeState, stepParams map[string]interface{},
	retryOption *models.RetryOption, correlationId string) ([]*models.NodeState, []*models.NodeState)

runStep runs one step of a batch

func (*FlowExecutor) SetFlowStatus

func (exec *FlowExecutor) SetFlowStatus(flow *models.Flow, status int) error

func (*FlowExecutor) Start

func (exec *FlowExecutor) Start(flow *models.Flow) error

Start starts an existing flow instance.

func (*FlowExecutor) Stop

func (exec *FlowExecutor) Stop(flow *models.Flow) error

Stop stopped a running flow by setting its status to SUCCESS.

type Job

type Job func() error

Job represents a task to be run by worker.

type Worker

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

Worker keep a queue to hold tasks, and run them one by one.

func NewWorker

func NewWorker(key int) *Worker

NewWorker creates a new worker.

func (*Worker) Start

func (w *Worker) Start()

Start starts the loop of the worker.

func (*Worker) Stop

func (w *Worker) Stop()

Stop stops the loop of the worker.

func (*Worker) Submit

func (w *Worker) Submit(f Job) error

Submit submits new job into queue of this worker, and return error if the queue if full.