pipeline

Documentation

Index

• Variables
• func NewNodeCtx(node Node) *nodeCtx
• func WrapErrRegDispather(err error) error
• type BaseNode
  • func NewBaseNode(name string, maxQueryLength int32) *BaseNode
  • func (node *BaseNode) MaxQueueLength() int32
  • func (node *BaseNode) Name() string
• type Msg
• type Node
• type Pipeline
• type StreamPipeline
  • func NewPipelineWithStream(dispatcher msgdispatcher.Client, nodeTtInterval time.Duration, ...) StreamPipeline

Variables

var (
	ErrEmptyPipeline     = errors.New("EmptyPipeline")
	ErrNilPosition       = errors.New("SeekToNilPosition")
	ErrRegisterDispather = errors.New("FailToRegisterDispather")
)

Functions

func NewNodeCtx

func NewNodeCtx(node Node) *nodeCtx

func WrapErrRegDispather

func WrapErrRegDispather(err error) error

Types

type BaseNode

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

func NewBaseNode

func NewBaseNode(name string, maxQueryLength int32) *BaseNode

func (*BaseNode) MaxQueueLength

func (node *BaseNode) MaxQueueLength() int32

length of pipeline input chnnel

func (*BaseNode) Name

func (node *BaseNode) Name() string

Return name of Node

type Msg

type Msg interface{}

type Node

type Node interface {
	Name() string
	MaxQueueLength() int32
	Operate(in Msg) Msg
}

type Pipeline

type Pipeline interface {
	Add(node ...Node)
	Start() error
	Close()
}

type StreamPipeline

type StreamPipeline interface {
	Pipeline
	ConsumeMsgStream(position *msgpb.MsgPosition) error
}

func NewPipelineWithStream

func NewPipelineWithStream(dispatcher msgdispatcher.Client, nodeTtInterval time.Duration, enableTtChecker bool, vChannel string) StreamPipeline