flow

Documentation

Index

• Variables
• type Context
  • func (ctx *Context) ExtendIn(data maps.Map) maps.Map
• type Flow
  • func Load(file string, id string) (*Flow, error)
  • func Select(name string) (*Flow, error)
  • func (flow *Flow) Exec(args ...interface{}) (interface{}, error)
  • func (flow *Flow) ExecNode(node *Node, ctx *Context, prev int) ([]interface{}, error)
  • func (flow *Flow) FormatResult(ctx *Context) (interface{}, error)
  • func (flow *Flow) Reload() (*Flow, error)
  • func (flow *Flow) RunProcess(node *Node, ctx *Context, data maps.Map) (interface{}, []interface{}, error)
  • func (flow *Flow) RunQuery(node *Node, ctx *Context, data maps.Map) (interface{}, []interface{}, error)
  • func (flow *Flow) WithGlobal(global map[string]interface{}) *Flow
  • func (flow *Flow) WithSID(sid string) *Flow
• type Node

Variables

var Flows = map[string]*Flow{}

Flows 已加载工作流列表

Types

type Context

type Context struct {
	In      []interface{}
	Res     map[string]interface{}
	Context *context.Context
	Cancel  context.CancelFunc
}

Context 工作流上下文

func (*Context) ExtendIn

func (ctx *Context) ExtendIn(data maps.Map) maps.Map

ExtendIn Extend params

type Flow

type Flow struct {
	ID          string                 `json:"-"`
	File        string                 `json:"-"`
	Name        string                 `json:"name"`
	Version     string                 `json:"version"`
	Description string                 `json:"description,omitempty"`
	Nodes       []Node                 `json:"nodes,omitempty"`
	Output      interface{}            `json:"output,omitempty"`
	Global      map[string]interface{} // 全局变量
	Sid         string                 // 会话ID
}

Flow 工作流

func Load

func Load(file string, id string) (*Flow, error)

Load the flow

func Select

func Select(name string) (*Flow, error)

Select 读取已加载Flow

func (*Flow) Exec

func (flow *Flow) Exec(args ...interface{}) (interface{}, error)

Exec execute flow

func (*Flow) ExecNode

func (flow *Flow) ExecNode(node *Node, ctx *Context, prev int) ([]interface{}, error)

ExecNode Execute node

func (*Flow) FormatResult

func (flow *Flow) FormatResult(ctx *Context) (interface{}, error)

FormatResult format result

func (*Flow) Reload

func (flow *Flow) Reload() (*Flow, error)

Reload 重新载入API

func (*Flow) RunProcess

func (flow *Flow) RunProcess(node *Node, ctx *Context, data maps.Map) (interface{}, []interface{}, error)

RunProcess exec process

func (*Flow) RunQuery

func (flow *Flow) RunQuery(node *Node, ctx *Context, data maps.Map) (interface{}, []interface{}, error)

RunQuery execute Query DSL

func (*Flow) WithGlobal

func (flow *Flow) WithGlobal(global map[string]interface{}) *Flow

WithGlobal 设定全局变量

func (*Flow) WithSID

func (flow *Flow) WithSID(sid string) *Flow

WithSID 设定会话ID

type Node

type Node struct {
	Name    string        `json:"name,omitempty"`
	Process string        `json:"process,omitempty"`
	Engine  string        `json:"engine,omitempty"` // 数据分析引擎名称
	Query   interface{}   `json:"query,omitempty"`  // 数据分析语言 Query Source
	DSL     share.DSL     `json:"-"`                // 数据分析语言 Query DSL
	Args    []interface{} `json:"args,omitempty"`
	Outs    []interface{} `json:"outs,omitempty"`
}

Node 工作流节点