flow

Documentation

Index

• Constants
• type Context
  • func (this *Context) GetAsString(k string) string
• type Engine
  • func (this *Engine) CommitTask(id int, success bool, msg string, next string, outParameters ...Parameter)
  • func (this *Engine) Init(id string, f *Flow)
  • func (this *Engine) Start()
• type Flow
  • func (this *Flow) AddTask(t *Task)
  • func (this *Flow) GetTask(tname string) *Task
  • func (this *Flow) LoadFromYamlFile(f string)
• type FlowManager
  • func (this *FlowManager) GetFlow(name string) *Flow
  • func (this *FlowManager) Init()
  • func (this *FlowManager) Publish(f *Flow) error
  • func (this *FlowManager) Remove(name string) error
• type InstanceStore
• type NotifyTaskStatus
• type Parameter
• type ParameterDefine
• type ParameterDefineType
• type TRetryLogic
• type TTaskStatus
• type TTaskType
  • func (this *TTaskType) UnmarshalYAML(unmarshal func(v interface{}) error) error
• type TTimeoutPolicy
• type Task
• type TaskDefine
• type TaskExecutor
• type TaskHandle
• type TaskInstance
• type WorkflowManager
  • func (this *WorkflowManager) AddTaskHandle(t TaskHandle)
  • func (this *WorkflowManager) Init()
  • func (this *WorkflowManager) Start(flowname string, parameters ...Parameter) error
• type WorkflowMetaManager

Constants

const (
	TS_Schedule  TTaskStatus = 1  //计划
	TS_Completed TTaskStatus = 2  //完成
	TS_Cancel    TTaskStatus = 4  //取消
	TS_Pause     TTaskStatus = 3  //暂停
	TS_Progress  TTaskStatus = 11 //进行中

	RL_fixed   TRetryLogic = 18 //重新安排任务
	RL_backoff TRetryLogic = 19 //重新安排
	//--------超时策略--------//
	TP_retry      TTimeoutPolicy = 21 //重试
	TP_timeout_wf TTimeoutPolicy = 22 //工作流标记为超时并终止
	TP_alert_only TTimeoutPolicy = 23 //仅仅作为警告

	//--------任务类型--------//
	TT_normal       TTaskType = 31 //标准任务
	TT_dynamic      TTaskType = 32 //动态任务
	TT_decide       TTaskType = 33 //判断决策任务
	TT_fork         TTaskType = 34 //并行任务
	TT_dynamic_fork TTaskType = 45 //动态分支并行
	TT_join         TTaskType = 46 //合并任务
	TT_wait         TTaskType = 47 //等待
	TT_http         TTaskType = 48 //http调用的任务
	TT_end          TTaskType = 0  //结束任务节点

)

Types

type Context

type Context map[string]interface{} //上下文

func (*Context) GetAsString

func (this *Context) GetAsString(k string) string

type Engine

type Engine struct {
	Id      string //实例唯一ID
	Name    string //对应的流程名称
	Version int    //对应的流程版本

	InstanceContext Context //流程实例的上下文
	// contains filtered or unexported fields
}

流程实例执行引擎

func (*Engine) CommitTask

func (this *Engine) CommitTask(id int, success bool, msg string, next string, outParameters ...Parameter)

更新任务状态

func (*Engine) Init

func (this *Engine) Init(id string, f *Flow)

func (*Engine) Start

func (this *Engine) Start()

开始执行流程 1、找到启动任务 2、执行启动任务

type Flow

type Flow struct {
	Name             string           //工作流程的名称，唯一
	Description      string           //流程描述
	Version          int              //版本号
	OutputParameters string           `yaml:"outputs"`
	InputParameters  string           `yaml:"inputs"` //输入参数列表，用于记录工作流所需要的输入，可选
	Tasks            map[string]*Task //任务定义
	StartTask        string           `yaml:"start"`
}

func (*Flow) AddTask

func (this *Flow) AddTask(t *Task)

func (*Flow) GetTask

func (this *Flow) GetTask(tname string) *Task

通过任务名称，获取task的定义

func (*Flow) LoadFromYamlFile

func (this *Flow) LoadFromYamlFile(f string)

type FlowManager

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

func (*FlowManager) GetFlow

func (this *FlowManager) GetFlow(name string) *Flow

func (*FlowManager) Init

func (this *FlowManager) Init()

func (*FlowManager) Publish

func (this *FlowManager) Publish(f *Flow) error

发布流程

func (*FlowManager) Remove

func (this *FlowManager) Remove(name string) error

删除流程

type InstanceStore

type InstanceStore interface {
}

状态存储

type NotifyTaskStatus

type NotifyTaskStatus func(flowid string, taskid int, success bool, err string, parameter ...Parameter)

type Parameter

type Parameter struct {
	Key   string
	Value interface{}
}

type ParameterDefine

type ParameterDefine struct {
	Key   string
	Type  ParameterDefineType
	Value string
}

type ParameterDefineType

type ParameterDefineType byte

const (
	PT_Const ParameterDefineType = 1 //常量
	PT_Refer ParameterDefineType = 2 //引用
)

type TRetryLogic

type TRetryLogic byte //重试逻辑

type TTaskStatus

type TTaskStatus byte //任务状态

type TTaskType

type TTaskType byte //任务类型

func (*TTaskType) UnmarshalYAML

func (this *TTaskType) UnmarshalYAML(unmarshal func(v interface{}) error) error

type TTimeoutPolicy

type TTimeoutPolicy byte //超时策略

type Task

type Task struct {
	Name            string    `yaml:"task"` //任务类型
	TaskName        string    `yaml:"code"` //任务名称，用于在流程中使用，必须唯一
	Type            TTaskType //任务类型
	Description     string    //任务描述
	Optional        bool      //是否可忽略，如果是true，当失败后，流程将继续.任务的状态反映为COMPLETED_WITH_ERRORS	默认为 false
	InputParameters Context   `yaml:"inputs"` //任务输入定义
	Expr            string    //表达式
	NextTask        []string  `yaml:"nexts"` //后续任务节点
}

任务

type TaskDefine

type TaskDefine struct {
	Name                   string            //任务类型，唯一
	RetryCount             int               //重试次数
	RetryLogic             TRetryLogic       //重试机制
	TimeoutSeconds         int64             //超时间 单位毫秒ms
	TimeoutPolicy          TTimeoutPolicy    //超时策略
	ResponseTimeoutSeconds int64             //返回超时时间
	InputParameters        []ParameterDefine //输入参数定义
	OutputKeys             []string          //任务输出字段
}

任务设置定义

type TaskExecutor

type TaskExecutor func(name string, flowid string, taskid int, parameter ...Parameter)

type TaskHandle

type TaskHandle interface {
	SetNotifylistener(l NotifyTaskStatus)
	GetName() string
	GetTaskDefine() TaskDefine
	Execute(flowid, taskid string, parameter ...Parameter) error
}

任务接口

type TaskInstance

type TaskInstance struct {
	FlowInstance    string  //流程的实例Id
	Id              int     //任务的实例Id
	TaskName        string  //任务名称
	InstanceContext Context //任务实例上下文，需要记录的对
	T               *Task
}

任务实例

type WorkflowManager

type WorkflowManager struct {
	FlowMeta WorkflowMetaManager
	Store    InstanceStore

	Prefix string
	// contains filtered or unexported fields
}

func (*WorkflowManager) AddTaskHandle

func (this *WorkflowManager) AddTaskHandle(t TaskHandle)

func (*WorkflowManager) Init

func (this *WorkflowManager) Init()

func (*WorkflowManager) Start

func (this *WorkflowManager) Start(flowname string, parameters ...Parameter) error

type WorkflowMetaManager

type WorkflowMetaManager interface {
	Publish(flow *Flow) error
	GetFlow(name string) *Flow
	Remove(name string) error
}

元信息