flow

Documentation

Index

• Variables
• func GetComponents() []*starriver.Component
• func LoadPipelineByJson(jsonConf string) (*starriver.PipelineConf, error)
• func LoadPipelineByYaml(yamlConf string) (*starriver.PipelineConf, error)
• func NewPipeline(conf starriver.PipelineConf) (starriver.Pipeline, error)
• func Rebuild(ctx context.Context, conf starriver.PipelineConf, ...) (starriver.DataContext, starriver.Pipeline, error)
• type Option
  • func SetEventHandler(eventHandler starriver.EventHandler) Option
• type RiverEngine
  • func NewRiverEngine(options ...Option) *RiverEngine
  • func (re *RiverEngine) CronRun(spec string, pipeline starriver.Pipeline, data map[string]interface{})
  • func (re *RiverEngine) Destroy()
  • func (re *RiverEngine) Run(dataContext starriver.DataContext, pipeline starriver.Pipeline) starriver.Result

Variables

var (
	NewDataContext = core.NewDataContext
	SetRequestID   = core.SetRequestID
	// NewSharedDataStore 新建自己的共享数据存储，主要用于修改默认的序列化方式。默认使用的 json 序列化方式对数字类型会有精度损失或类型错乱。
	NewSharedDataStore = builtin.NewSharedDataStore
	SetSharedDataStore = core.SetSharedDataStore
	SetLogger          = core.SetLogger
	SetLogLevel        = core.SetLogLevel
)

Functions

func GetComponents

func GetComponents() []*starriver.Component

func LoadPipelineByJson

func LoadPipelineByJson(jsonConf string) (*starriver.PipelineConf, error)

func LoadPipelineByYaml

func LoadPipelineByYaml(yamlConf string) (*starriver.PipelineConf, error)

func NewPipeline

func NewPipeline(conf starriver.PipelineConf) (starriver.Pipeline, error)

func Rebuild

func Rebuild(ctx context.Context, conf starriver.PipelineConf, taskStatuses map[string]starriver.TaskStatus,
	snapshot starriver.SharedDataStore, initialData map[string]interface{},
	opts ...core.ContextOption) (starriver.DataContext, starriver.Pipeline, error)

Rebuild a pipeline from a snapshot

Types

type Option

type Option func(*RiverEngine)

func SetEventHandler

func SetEventHandler(eventHandler starriver.EventHandler) Option

type RiverEngine

type RiverEngine struct {
	WorkerConcurrency int
	Semaphore         util.Semaphore
	LoggingEnabled    bool
	DebugEnabled      bool
	EventHandler      starriver.EventHandler
	// contains filtered or unexported fields
}

func NewRiverEngine

func NewRiverEngine(options ...Option) *RiverEngine

NewRiverEngine new a river engine

func (*RiverEngine) CronRun

func (re *RiverEngine) CronRun(spec string, pipeline starriver.Pipeline, data map[string]interface{})

func (*RiverEngine) Destroy

func (re *RiverEngine) Destroy()

func (*RiverEngine) Run

func (re *RiverEngine) Run(dataContext starriver.DataContext, pipeline starriver.Pipeline) starriver.Result