flow

Documentation

Index

• Variables
• func GoAndRun(ctx context.Context, wg *sync.WaitGroup, reg *NodesReg, firstValue []byte)
• func PrepareAllInputs(inputs NodeConnection) []string
• func PrepareOutputs(outputs NodeConnection) [][]Connection
• func VisitAndFetch(ctx context.Context, reg *NodesReg) error
• type Connection
• type Connections
• type ContextTagsValue
• type ContextType
• type CountStucker
• type Inputs
  • func PrepareInputs(inputs NodeConnection) []Inputs
• type NodeAttachments
• type NodeConnection
• type NodeData
• type NodeDirectGo
• type NodeRet
• type NodeRetDatas
• type NodeRetRespond
• type NodeRetRespondData
• type NodeRetSelection
• type NodeRetValues
• type Noder
• type NoderEndpoint
• type NodesData
  • func ParseData(content []byte) (NodesData, error)
• type NodesReg
  • func DataToNode(ctx context.Context, controlName, startName, method string, datas NodesData, ...) (*NodesReg, error)
  • func NewNodesReg(controlName, startName, method string, appStore *registry.Registry) *NodesReg
  • func StartFlow(ctx context.Context, wg *sync.WaitGroup, controlName, endPoint, method string, ...) (*NodesReg, error)
  • func (r *NodesReg) AddCleanup(v func())
  • func (r *NodesReg) AddError(err error)
  • func (r *NodesReg) AddStuctCancel(v context.CancelFunc)
  • func (r *NodesReg) CancelStucks()
  • func (r *NodesReg) Clear()
  • func (r *NodesReg) Get(number string) (Noder, bool)
  • func (r *NodesReg) GetChan() <-chan Respond
  • func (r *NodesReg) GetStarts() []Starts
  • func (r *NodesReg) GetStuctCancel(ctx context.Context) context.Context
  • func (r *NodesReg) Set(number string, node Noder)
  • func (r *NodesReg) SetChanInactive()
  • func (r *NodesReg) UpdateStuck(typeCount CountStucker, trigger bool)
• type Respond
• type Starts

Variables

var (
	Input1 = "input_1"
	Input2 = "input_2"
)

var (
	ErrStopGoroutine    = errors.New("stop goroutine")
	ErrEndpointNotFound = errors.New("endpoint not found")
)

var NodeTypes = make(map[string]func(context.Context, *NodesReg, NodeData, string) (Noder, error))

NodeTypes hold new function of releated node.

Functions

func GoAndRun

func GoAndRun(ctx context.Context, wg *sync.WaitGroup, reg *NodesReg, firstValue []byte)

func PrepareAllInputs

func PrepareAllInputs(inputs NodeConnection) []string

func PrepareOutputs

func PrepareOutputs(outputs NodeConnection) [][]Connection

func VisitAndFetch

func VisitAndFetch(ctx context.Context, reg *NodesReg) error

Types

type Connection

type Connection = struct {
	Node   string `json:"node"`
	Output string `json:"output"`
}

type Connections

type Connections = struct {
	Connections []Connection `json:"connections"`
}

type ContextTagsValue

type ContextTagsValue = map[string]struct{}

type ContextType

type ContextType string

const CtxTags ContextType = "tags"

type CountStucker

type CountStucker uint

const (
	CountTotalIncrease CountStucker = iota + 1
	CountTotalDecrease
	CountStuckIncrease
	CountStuckDecrease
)

type Inputs

type Inputs struct {
	Node      string
	InputName string
	Active    bool
}

func PrepareInputs

func PrepareInputs(inputs NodeConnection) []Inputs

type NodeAttachments

type NodeAttachments interface {
	GetAttachments() []email.Attach
}

type NodeConnection

type NodeConnection = map[string]Connections

type NodeData

type NodeData struct {
	Data    map[string]interface{} `json:"data"`
	Inputs  NodeConnection         `json:"inputs"`
	Outputs NodeConnection         `json:"outputs"`
	Name    string
}

type NodeDirectGo

type NodeDirectGo interface {
	IsDirectGo() NodeRet
}

NodeDirectGo for direct go.

type NodeRet

type NodeRet interface {
	GetBinaryData() []byte
}

NodeRet simple return.

type NodeRetDatas

type NodeRetDatas interface {
	GetBinaryDatas() [][]byte
}

NodeRetDatas usuful for-loop operation.

type NodeRetRespond

type NodeRetRespond interface {
	GetRespond() Respond
}

NodeRetRespond using for responding request.

type NodeRetRespondData

type NodeRetRespondData interface {
	GetRespondData() Respond
}

NodeRetRespondData using if next node wants to respond data.

type NodeRetSelection

type NodeRetSelection interface {
	GetSelection() []int
}

NodeRetSelection usable if more than one output and want to choice between of them. Write to output numbers 0-4-5.

type NodeRetValues

type NodeRetValues interface {
	GetBinaryValues() []byte
}

type Noder

type Noder interface {
	GetType() string
	Run(context.Context, *sync.WaitGroup, *registry.Registry, NodeRet, string) (NodeRet, error)
	Fetch(context.Context, *gorm.DB) error
	IsFetched() bool
	Validate(context.Context) error
	ActiveInput(string, map[string]struct{})
	Next(int) []Connection
	NextCount() int
	IsRespond() bool
	Check()
	IsChecked() bool
	IsDisabled() bool
	NodeID() string
	Tags() []string
}

Noder for nodes like script, endpoint.

type NoderEndpoint

type NoderEndpoint interface {
	Endpoint() string
	Methods() []string
	Tags() []string
}

type NodesData

type NodesData = map[string]NodeData

NodesData is content's represent.

func ParseData

func ParseData(content []byte) (NodesData, error)

type NodesReg

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

NodesReg hold concreate information of nodes and start points.

func DataToNode

func DataToNode(
	ctx context.Context,
	controlName, startName, method string,
	datas NodesData,
	appStore *registry.Registry,
) (*NodesReg, error)

func NewNodesReg

func NewNodesReg(controlName, startName, method string, appStore *registry.Registry) *NodesReg

func StartFlow

func StartFlow(
	ctx context.Context,
	wg *sync.WaitGroup,
	controlName, endPoint, method string,
	content []byte,
	appStore *registry.Registry,
	value []byte,
) (*NodesReg, error)

func (*NodesReg) AddCleanup

func (r *NodesReg) AddCleanup(v func())

func (*NodesReg) AddError

func (r *NodesReg) AddError(err error)

func (*NodesReg) AddStuctCancel

func (r *NodesReg) AddStuctCancel(v context.CancelFunc)

func (*NodesReg) CancelStucks

func (r *NodesReg) CancelStucks()

CancelStucks cancel all stuck context of nodes and clear stuck context list.

func (*NodesReg) Clear

func (r *NodesReg) Clear()

Clear clear all context.

func (*NodesReg) Get

func (r *NodesReg) Get(number string) (Noder, bool)

func (*NodesReg) GetChan

func (r *NodesReg) GetChan() <-chan Respond

func (*NodesReg) GetStarts

func (r *NodesReg) GetStarts() []Starts

func (*NodesReg) GetStuctCancel

func (r *NodesReg) GetStuctCancel(ctx context.Context) context.Context

func (*NodesReg) Set

func (r *NodesReg) Set(number string, node Noder)

Set a concreate node to registry. Number is a node number like 2, 4.

func (*NodesReg) SetChanInactive

func (r *NodesReg) SetChanInactive()

func (*NodesReg) UpdateStuck

func (r *NodesReg) UpdateStuck(typeCount CountStucker, trigger bool)

type Respond

type Respond struct {
	Header   map[string]interface{} `json:"header"`
	FileName string                 `json:"-"`
	Data     []byte                 `json:"data"`
	Status   int                    `json:"status"`
	IsError  bool                   `json:"-"`
}

type Starts

type Starts = struct {
	Connection
	Tags map[string]struct{}
}