v1

Documentation

Index

• Variables
• type Context
• type Dag
  • func NewDag() *Dag
  • func (currentDag *Dag) Append(dag *Dag)
  • func (currentDag *Dag) ConditionalBranch(vertex string, conditions []string, condition sdk.Condition, options ...Option) (conditionDags map[string]*Dag)
  • func (currentDag *Dag) Definition() (*sdk.DagExporter, error)
  • func (currentDag *Dag) DefinitionJson() ([]byte, error)
  • func (currentDag *Dag) Edge(from, to string, opts ...Option)
  • func (currentDag *Dag) ForEachBranch(vertex string, foreach sdk.ForEach, options ...Option) (dag *Dag)
  • func (currentDag *Dag) Node(vertex string, workload operation.Modifier, options ...Option) *Node
  • func (currentDag *Dag) SubDag(vertex string, dag *Dag)
  • func (currentDag *Dag) Validate() error
• type DataStore
• type ExecutionOptions
• type Node
• type Option
  • func Aggregator(aggregator sdk.Aggregator) Option
  • func Forwarder(forwarder sdk.Forwarder) Option
  • func InvokeEdge() Option
  • func OnFailure(handler operation.FuncErrorHandler) Option
• type StateStore
• type Workflow
  • func GetWorkflow(pipeline *sdk.Pipeline) *Workflow
  • func (flow *Workflow) Dag() *Dag
  • func (flow *Workflow) Finally(handler sdk.PipelineHandler)
  • func (flow *Workflow) GetPipeline() *sdk.Pipeline
  • func (flow *Workflow) OnFailure(handler sdk.PipelineErrorHandler)
  • func (flow *Workflow) SetDag(dag *Dag)

Variables

var (
	// Execution specify a edge doesn't forwards a data
	// but rather mention a execution direction
	Execution = InvokeEdge()
)

Types

type Context

type Context sdk.Context

type Dag

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

func NewDag

func NewDag() *Dag

NewDag creates a new dag separately from pipeline

func (*Dag) Append

func (currentDag *Dag) Append(dag *Dag)

Append generalizes a separate dag by appending its properties into current dag. Provided dag should be mutually exclusive

func (*Dag) ConditionalBranch

func (currentDag *Dag) ConditionalBranch(vertex string, conditions []string, condition sdk.Condition,
	options ...Option) (conditionDags map[string]*Dag)

ConditionalBranch composites multiple dags as a sub-dag which executes for each conditions returned by the Condition function dynamically It returns the set of dags based on the set of condition passed

func (*Dag) Definition

func (currentDag *Dag) Definition() (*sdk.DagExporter, error)

func (*Dag) DefinitionJson

func (currentDag *Dag) DefinitionJson() ([]byte, error)

func (*Dag) Edge

func (currentDag *Dag) Edge(from, to string, opts ...Option)

Edge adds a directed edge between two vertex as <from>-><to>

func (*Dag) ForEachBranch

func (currentDag *Dag) ForEachBranch(vertex string, foreach sdk.ForEach, options ...Option) (dag *Dag)

ForEachBranch composites a sub-dag which executes for each value returned by ForEach function dynamically It returns the sub-dag that will be executed for each value

func (*Dag) Node

func (currentDag *Dag) Node(vertex string, workload operation.Modifier, options ...Option) *Node

Node adds a new vertex by id

func (*Dag) SubDag

func (currentDag *Dag) SubDag(vertex string, dag *Dag)

SubDag composites a separate dag as a node.

func (*Dag) Validate

func (currentDag *Dag) Validate() error

type DataStore

type DataStore sdk.DataStore

type ExecutionOptions

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

ExecutionOptions options for branching in DAG

type Node

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

type Option

type Option func(*ExecutionOptions)

func Aggregator

func Aggregator(aggregator sdk.Aggregator) Option

Aggregator aggregates all outputs into one

func Forwarder

func Forwarder(forwarder sdk.Forwarder) Option

Forwarder encodes request based on need for children vertex by default the data gets forwarded as it is

func InvokeEdge

func InvokeEdge() Option

InvokeEdge denotes a edge doesn't forwards a data, but rather provides only an execution flow

func OnFailure

func OnFailure(handler operation.FuncErrorHandler) Option

OnFailure Specify a function failure handler

type StateStore

type StateStore sdk.StateStore

type Workflow

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

func GetWorkflow

func GetWorkflow(pipeline *sdk.Pipeline) *Workflow

GetWorkflow initiates a flow with a pipeline

func (*Workflow) Dag

func (flow *Workflow) Dag() *Dag

Dag provides the workflow dag object

func (*Workflow) Finally

func (flow *Workflow) Finally(handler sdk.PipelineHandler)

Finally sets an execution finish handler routine it will be called once the execution has finished with state either Success/Failure

func (*Workflow) GetPipeline

func (flow *Workflow) GetPipeline() *sdk.Pipeline

GetPipeline expose the underlying pipeline object

func (*Workflow) OnFailure

func (flow *Workflow) OnFailure(handler sdk.PipelineErrorHandler)

OnFailure set a failure handler routine for the pipeline

func (*Workflow) SetDag

func (flow *Workflow) SetDag(dag *Dag)

SetDag apply a predefined dag, and override the default dag