workflow

Documentation

Index

• Variables
• func ExecutePredicate(context BaseState, predicate Predicate, dep Dependency) (bool, error)
• func ExecuteReshaper(context BaseState, reshaper Reshaper) (schema.Schema, error)
• func GetRunID(state State) string
• func NewMachine(di Dependency, state State) *machine.Machine[Command, State]
• func ToStrExpr(expr Expr, depth int) string
• func ToStrPredicate(predicate Predicate, depth int) string
• func ToStrReshaper(reshaper Reshaper, depth int) string
• func ToStrSchema(x schema.Schema, depth int) string
• func ToStrWorkflow(workflow Worflow, depth int) string
• type And
• type Apply
• type ApplyAwaitOptions
• type Assign
• type Await
• type BaseState
  • func GetBaseState(status State) BaseState
• type Callback
• type Choose
• type Compare
• type DI
  • func (di *DI) FindFunction(funcID string) (Function, error)
  • func (di *DI) FindWorkflow(flowID string) (*Flow, error)
  • func (di *DI) GenerateCallbackID() string
  • func (di *DI) GenerateRunID() string
  • func (di *DI) MaxRetries() int64
  • func (di *DI) TimeNow() time.Time
• type DelayRun
• type Dependency
• type Done
• type End
• type Error
• type Execution
• type Flow
• type FlowRef
• type Function
• type FunctionInput
• type FunctionOutput
• type GetValue
• type NextOperation
• type Not
• type Or
• type ResumeOptions
• type ResumeSchedule
• type Run
  • func ScheduleNext(x State, dep Dependency) *Run
• type ScheduleRun
• type ScheduleStopped
• type Scheduled
• type SetValue
• type StopSchedule
• type TryRecover

Variables

var (
	ErrAlreadyStarted         = errors.New("already started")
	ErrCallbackNotMatch       = errors.New("callback not match")
	ErrInvalidStateTransition = errors.New("invalid state transition")
	ErrExpressionHasResult    = errors.New("expression has result")
	ErrStateReachEnd          = errors.New("cannot apply commands, when workflow is completed")
	ErrMaxRetriesReached      = errors.New("max retries reached")
	ErrFlowNotFound           = errors.New("flow not found")
	ErrFlowNotSet             = errors.New("flow not set")
	ErrIntervalParse          = errors.New("failed to parse interval")
	ErrRunIDNotMatch          = errors.New("run id not match")
)

Functions

func ExecutePredicate

func ExecutePredicate(context BaseState, predicate Predicate, dep Dependency) (bool, error)

func ExecuteReshaper

func ExecuteReshaper(context BaseState, reshaper Reshaper) (schema.Schema, error)

func GetRunID

func GetRunID(state State) string

func NewMachine

func NewMachine(di Dependency, state State) *machine.Machine[Command, State]

func ToStrExpr

func ToStrExpr(expr Expr, depth int) string

func ToStrPredicate

func ToStrPredicate(predicate Predicate, depth int) string

func ToStrReshaper

func ToStrReshaper(reshaper Reshaper, depth int) string

func ToStrSchema

func ToStrSchema(x schema.Schema, depth int) string

func ToStrWorkflow

func ToStrWorkflow(workflow Worflow, depth int) string

ToStrWorkflow returns string representation of workflow AST, a string is a meta program code, similar to go code, just declarative. Example:

func FlowHelloWorld(input string) string {
	var res string
	res = concat("hello ", input)
	return res
}

Types

type And

type And struct {
	L []Predicate
}

type Apply

type Apply struct {
	ID    string
	Name  string
	Args  []Reshaper
	Await *ApplyAwaitOptions
}

type ApplyAwaitOptions

type ApplyAwaitOptions struct {
	Timeout int64
}

type Assign

type Assign struct {
	ID    string
	VarOk string
	// if VarErr is not empty, then error will be assigned to this variable
	// to give chance to handle it, before it will be returned to the caller
	// otherwise, any error will stop execution of the program
	VarErr string
	Val    Expr
}

type Await

type Await struct {
	CallbackID string
	Timeout    int64
	BaseState  BaseState
}

type BaseState

type BaseState struct {
	Flow       Worflow // Flow is a reference to the flow that describes execution
	RunID      string  // RunID is a unique identifier of the execution
	StepID     string  // StepID is a unique identifier of the step in the execution
	Variables  map[string]schema.Schema
	ExprResult map[string]schema.Schema

	// Default values
	DefaultMaxRetries int64

	RunOption RunOption
}

func GetBaseState

func GetBaseState(status State) BaseState

type Callback

type Callback struct {
	CallbackID string
	Result     schema.Schema
}

type Choose

type Choose struct {
	ID   string
	If   Predicate
	Then []Expr
	Else []Expr
}

type Compare

type Compare struct {
	Operation string
	Left      Reshaper
	Right     Reshaper
}

type DI

type DI struct {
	FindFunctionF       func(funcID string) (Function, error)
	FindWorkflowF       func(flowID string) (*Flow, error)
	GenerateCallbackIDF func() string
	GenerateRunIDF      func() string

	// Defaults
	DefaultMaxRetries int64
	MockTimeNow       *time.Time
}

func (*DI) FindFunction

func (di *DI) FindFunction(funcID string) (Function, error)

func (*DI) FindWorkflow

func (di *DI) FindWorkflow(flowID string) (*Flow, error)

func (*DI) GenerateCallbackID

func (di *DI) GenerateCallbackID() string

func (*DI) GenerateRunID

func (di *DI) GenerateRunID() string

func (*DI) MaxRetries

func (di *DI) MaxRetries() int64

func (*DI) TimeNow

func (di *DI) TimeNow() time.Time

type DelayRun

type DelayRun struct {
	// DelayBySeconds
	DelayBySeconds int64
}

type Dependency

type Dependency interface {
	FindWorkflow(flowID string) (*Flow, error)
	FindFunction(funcID string) (Function, error)
	GenerateCallbackID() string
	GenerateRunID() string
	MaxRetries() int64
	TimeNow() time.Time
}

type Done

type Done struct {
	Result    schema.Schema
	BaseState BaseState
}

type End

type End struct {
	ID     string
	Result Reshaper
}

type Error

type Error struct {
	Code    string
	Reason  string
	Retried int64
	//MaxRetries int64
	BaseState BaseState
}

type Execution

type Execution struct {
	FlowID    string
	Status    State
	Location  string
	StartTime int64
	EndTime   int64
	Variables map[string]schema.Schema
}

type Flow

type Flow struct {
	Name string // name of the flow
	Arg  string // name of the argument, which will hold the input to the flow
	Body []Expr
}

type FlowRef

type FlowRef struct {
	FlowID string
}

type Function

type Function func(args *FunctionInput) (*FunctionOutput, error)

type FunctionInput

type FunctionInput struct {
	// Name acts as unique function ID
	Name string
	// CallbackID is used to identify callback function, and when its set
	// it means that function is async, and should return result by calling callback endpoint with CallbackID
	CallbackID string
	Args       []schema.Schema
}

FunctionDef struct {
	Name string
	Input schema.ShapeDef
	Output schema.ShapeDef
}

type FunctionOutput

type FunctionOutput struct {
	Result schema.Schema
}

type GetValue

type GetValue struct {
	Path string
}

type NextOperation

type NextOperation struct {
	Result    schema.Schema
	BaseState BaseState
}

type Not

type Not struct {
	P Predicate
}

type Or

type Or struct {
	L []Predicate
}

type ResumeOptions

type ResumeOptions struct {
	Timeout int64
}

type ResumeSchedule

type ResumeSchedule struct {
	ParentRunID string
}

type Run

type Run struct {
	Flow  Worflow
	Input schema.Schema
	// Schedule run
	RunOption RunOption
}

func ScheduleNext

func ScheduleNext(x State, dep Dependency) *Run

type ScheduleRun

type ScheduleRun struct {
	// CRON like definition
	Interval string
}

type ScheduleStopped

type ScheduleStopped struct {
	ParentRunID string
	BaseState   BaseState
}

type Scheduled

type Scheduled struct {
	// ExpectedRunTimestamp is server timestamp + DelayBySeconds
	ExpectedRunTimestamp int64
	// ParentRunID is a reference to the original run, that scheduled this run and any between
	// ParentRunID is used to track history of the execution, and is stable reference to the original run
	ParentRunID string
	BaseState   BaseState
}

Scheduled is a state that is used to schedule execution of the flow, once or periodically

type SetValue

type SetValue struct {
	Value schema.Schema
}

type StopSchedule

type StopSchedule struct {
	// ParentRunID can be stopped by user, or by system
	ParentRunID string
}

type TryRecover

type TryRecover struct{}