flow

Documentation

Index

• Constants
• type Base
• type Flow
  • func NewFlow(storage *StorageOption, opts ...FlowOption) *Flow
  • func (f *Flow) DeletePipeline(id string) error
  • func (f *Flow) GetPipelineByID(id string, needNode, needPipelineRun bool) (*Pipeline, error)
  • func (f *Flow) NewPipeline(args *PipelineArgs) error
  • func (f *Flow) NewPipelineRun(args *PipelineRunArgs) error
  • func (f *Flow) UpdatePipelineRemark(id, remark string) error
  • func (f *Flow) UpdatePipelineTopic(id string, topic string) error
• type FlowOption
  • func WithContext(ctx context.Context) FlowOption
  • func WithDelay(d time.Duration) FlowOption
• type MysqlOption
• type Node
• type NodeArgs
• type NodeData
• type Pipeline
• type PipelineArgs
• type PipelineRun
• type PipelineRunArgs
• type RedisOption
• type StorageOption

Constants

const (
	PipelineRunStatusProcessing = "processing"
	PipelineRunStatusPending    = "pending"
	PipelineRunStatusSuccess    = "success"
	PipelineRunStatusFailed     = "failed"
)

Types

type Base

type Base struct {
	ID        string `gorm:"primaryKey;"`
	CreatedAt int64
	UpdatedAt int64
	DeletedAt gorm.DeletedAt `gorm:"index"`
}

type Flow

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

func NewFlow

func NewFlow(storage *StorageOption, opts ...FlowOption) *Flow

func (*Flow) DeletePipeline

func (f *Flow) DeletePipeline(id string) error

DeletePipeline will delete the pipeline and all nodes in it.

func (*Flow) GetPipelineByID

func (f *Flow) GetPipelineByID(id string, needNode, needPipelineRun bool) (*Pipeline, error)

func (*Flow) NewPipeline

func (f *Flow) NewPipeline(args *PipelineArgs) error

func (*Flow) NewPipelineRun

func (f *Flow) NewPipelineRun(args *PipelineRunArgs) error

func (*Flow) UpdatePipelineRemark

func (f *Flow) UpdatePipelineRemark(id, remark string) error

func (*Flow) UpdatePipelineTopic

func (f *Flow) UpdatePipelineTopic(id string, topic string) error

type FlowOption

type FlowOption func(*Flow)

func WithContext

func WithContext(ctx context.Context) FlowOption

func WithDelay

func WithDelay(d time.Duration) FlowOption

type MysqlOption

type MysqlOption struct {
	Host     string
	Port     string
	User     string
	Password string
	DbName   string
}

type Node

type Node struct {
	Base
	Name       string     `json:"name"`
	Sequence   int        `gorm:"index:idx_seq_pipe" json:"sequence"`
	PipelineID string     `gorm:"index:idx_seq_pipe" json:"pipeline_id"`
	PrevNodeID string     `json:"prev_node_id"`
	NextNodeID string     `json:"next_node_id"`
	Template   string     `json:"template"`
	Pipeline   *Pipeline  `json:"pipeline"`
	NodeData   []NodeData `json:"node_data"`
}

type NodeArgs

type NodeArgs struct {
	Name       string
	PipelineID string `json:"pipeline_id"`
	Template   string `json:"template"`
}

type NodeData

type NodeData struct {
	Base
	PipelineRunID string       `gorm:"index:idx_run_node" json:"pipeline_run_id"`
	NodeID        string       `gorm:"index:idx_run_node" json:"node_id"`
	Data          string       `json:"data"`
	Node          *Node        `json:"node"`
	PipelineRun   *PipelineRun `json:"pipeline_run"`
}

type Pipeline

type Pipeline struct {
	Base
	Topic       string        `json:"topic"`
	Head        string        `json:"head"` // head node ID
	Remark      string        `json:"remark"`
	Node        []Node        `json:"node"`
	PipelineRun []PipelineRun `json:"pipeline_run"`
}

type PipelineArgs

type PipelineArgs struct {
	Topic  string
	Remark string
	Nodes  []NodeArgs
}

type PipelineRun

type PipelineRun struct {
	Base
	PipelineID string     `gorm:"index:idx_pipe_stage" json:"pipeline_id"`
	Stage      string     `gorm:"idnex:idx_pipe_stage" json:"stage"` //node ID
	Status     string     `json:"status"`                            // processing, pending, success, failed
	Pipeline   *Pipeline  `json:"pipeline"`
	NodeData   []NodeData `json:"node_data"`
}

type PipelineRunArgs

type PipelineRunArgs struct {
	PipelineID string
}

type RedisOption

type RedisOption struct {
	Host     string
	Port     string
	Password string
	Db       int
}

type StorageOption

type StorageOption struct {
	Mysql *MysqlOption
	Redis *RedisOption
}