distribution

Documentation

Index

• Constants
• func DataToFlowError(data []byte) (*pluginapi.FlowError, error)
• func DataToFlowStop(data []byte) (*pluginapi.FlowStop, error)
• func DataToModel(data []byte) (pluginapi.Model, error)
• func FlowErrorToData(flowError *pluginapi.FlowError) ([]byte, error)
• func FlowStopToData(flowStop *pluginapi.FlowStop) ([]byte, error)
• func ModelToData(model pluginapi.Model) ([]byte, error)
• func NewLocalFlowInvoker() pluginapi.FlowInvoker
• func NewNatsFlowInvoker(appName, addresses string) pluginapi.FlowInvoker
• func NewSingleDispatchDecider() pluginapi.DispatchDecider
• func TransferModel(src, dst pluginapi.Model) error
• type AbstractDispatchDecider
• type LocalFlowInvoker
  • func (l *LocalFlowInvoker) AddPipeline(pipelineName string, process pluginapi.PipelineProcess) error
  • func (l *LocalFlowInvoker) Invoke(pipelineFullName string, model pluginapi.Model) error
  • func (l *LocalFlowInvoker) Metadata() pluginapi.FlowInvokerMeta
  • func (l *LocalFlowInvoker) StartFlowInvoker() error
  • func (l *LocalFlowInvoker) StopFlowInvoker() error
• type NatsFlowInvoker
  • func (n *NatsFlowInvoker) AddPipeline(pipelineName string, process pluginapi.PipelineProcess) error
  • func (n *NatsFlowInvoker) Invoke(pipelineFullName string, model pluginapi.Model) error
  • func (n *NatsFlowInvoker) Metadata() pluginapi.FlowInvokerMeta
  • func (n *NatsFlowInvoker) StartFlowInvoker() error
  • func (n *NatsFlowInvoker) StopFlowInvoker() error
• type SingleDispatchDecider
  • func (s *SingleDispatchDecider) AddFlowInvoker(f pluginapi.FlowInvoker) error
  • func (s *SingleDispatchDecider) InjectLocalPipeline(pipelineFullName string, process pluginapi.PipelineProcess) error
  • func (s *SingleDispatchDecider) PipelineDispatcher(pipelineFullName string) pluginapi.PipelineProcess
  • func (s *SingleDispatchDecider) StartDispatcher() error
  • func (s *SingleDispatchDecider) StopDispatcher() error

Constants

const (
	NatsFlowStopStatusCode  = "9998"
	NatsFlowErrorStatusCode = "9999"
)

Functions

func DataToFlowError

func DataToFlowError(data []byte) (*pluginapi.FlowError, error)

func DataToFlowStop

func DataToFlowStop(data []byte) (*pluginapi.FlowStop, error)

func DataToModel

func DataToModel(data []byte) (pluginapi.Model, error)

func FlowErrorToData

func FlowErrorToData(flowError *pluginapi.FlowError) ([]byte, error)

func FlowStopToData

func FlowStopToData(flowStop *pluginapi.FlowStop) ([]byte, error)

func ModelToData

func ModelToData(model pluginapi.Model) ([]byte, error)

func NewLocalFlowInvoker

func NewLocalFlowInvoker() pluginapi.FlowInvoker

func NewNatsFlowInvoker

func NewNatsFlowInvoker(appName, addresses string) pluginapi.FlowInvoker

func NewSingleDispatchDecider

func NewSingleDispatchDecider() pluginapi.DispatchDecider

func TransferModel

func TransferModel(src, dst pluginapi.Model) error

Types

type AbstractDispatchDecider

type AbstractDispatchDecider struct {
}

AbstractDispatchDecider is used for dispatching request to next step A decider should make decisions based on the considerations of: * Interaction types like rpc/event * Keep minimum request/response serialization cost * Burden of communication or latency * Other resource concerns When enabling distributed processing model, multiple nodes will be working as: * Dispatcher mode: entrypoint node(usually the source connector node) dispatches each stage of the pipeline to different nodes * Relay/SEDA mode: the request will be transferred one by one node in the pipeline and finally return to the entrypoint node to respond result(if required) * Note: choosing mode should have the following considerations:

• passing full context
• communication cost
• require more on serialization
• other resource concerns

type LocalFlowInvoker

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

func (*LocalFlowInvoker) AddPipeline

func (l *LocalFlowInvoker) AddPipeline(pipelineName string, process pluginapi.PipelineProcess) error

func (*LocalFlowInvoker) Invoke

func (l *LocalFlowInvoker) Invoke(pipelineFullName string, model pluginapi.Model) error

func (*LocalFlowInvoker) Metadata

func (l *LocalFlowInvoker) Metadata() pluginapi.FlowInvokerMeta

func (*LocalFlowInvoker) StartFlowInvoker

func (l *LocalFlowInvoker) StartFlowInvoker() error

func (*LocalFlowInvoker) StopFlowInvoker

func (l *LocalFlowInvoker) StopFlowInvoker() error

type NatsFlowInvoker

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

func (*NatsFlowInvoker) AddPipeline

func (n *NatsFlowInvoker) AddPipeline(pipelineName string, process pluginapi.PipelineProcess) error

func (*NatsFlowInvoker) Invoke

func (n *NatsFlowInvoker) Invoke(pipelineFullName string, model pluginapi.Model) error

func (*NatsFlowInvoker) Metadata

func (n *NatsFlowInvoker) Metadata() pluginapi.FlowInvokerMeta

func (*NatsFlowInvoker) StartFlowInvoker

func (n *NatsFlowInvoker) StartFlowInvoker() error

func (*NatsFlowInvoker) StopFlowInvoker

func (n *NatsFlowInvoker) StopFlowInvoker() error

type SingleDispatchDecider

type SingleDispatchDecider struct {
	AbstractDispatchDecider
	// contains filtered or unexported fields
}

func (*SingleDispatchDecider) AddFlowInvoker

func (s *SingleDispatchDecider) AddFlowInvoker(f pluginapi.FlowInvoker) error

func (*SingleDispatchDecider) InjectLocalPipeline

func (s *SingleDispatchDecider) InjectLocalPipeline(pipelineFullName string, process pluginapi.PipelineProcess) error

func (*SingleDispatchDecider) PipelineDispatcher

func (s *SingleDispatchDecider) PipelineDispatcher(pipelineFullName string) pluginapi.PipelineProcess

func (*SingleDispatchDecider) StartDispatcher

func (s *SingleDispatchDecider) StartDispatcher() error

func (*SingleDispatchDecider) StopDispatcher

func (s *SingleDispatchDecider) StopDispatcher() error