stream

Documentation

Index

• type ProcessorWrapper
  • func NewProcessorWrapper(pluginType processors.ProcessorDriver, config interface{}, ...) ProcessorWrapper
  • func (p *ProcessorWrapper) EvolveSchema(s *schema.StreamSchemaObj) error
  • func (p *ProcessorWrapper) LoadDriver(driver processors.ProcessorDriver, cfg interface{}) (processors.DataProcessor, error)
  • func (p *ProcessorWrapper) Process(msg *message.Message) (*message.Message, error)
  • func (p *ProcessorWrapper) SetStreamContext(ctx *stream_context.Context)
• type SinkWrapper
  • func NewSinkWrapper(pluginType sinks.SinkDriver, config config.Configuration, ...) SinkWrapper
  • func (p *SinkWrapper) Init() error
  • func (p *SinkWrapper) LoadDriver(driver sinks.SinkDriver, cfg config.Configuration) sinks.DataSink
  • func (p *SinkWrapper) SetExpectedSchema(s *schema.StreamSchemaObj)
  • func (p *SinkWrapper) SetStreamContext(ctx *stream_context.Context)
  • func (p *SinkWrapper) Write(msg *message.Message) error
• type SourceWrapper
  • func NewSourceWrapper(pluginType sources.SourceDriver, config config.Configuration) SourceWrapper
  • func (p *SourceWrapper) Events() chan sources.MessageEvent
  • func (p *SourceWrapper) GetPluginConfigs(driver sources.SourceDriver, fcg *yaml.Node) (any, error)
  • func (p *SourceWrapper) Init(appctx *stream_context.Context) error
  • func (p *SourceWrapper) LoadDriver(driver sources.SourceDriver, fcg config.Configuration) sources.DataSource
  • func (p *SourceWrapper) SetStreamContext(ctx *stream_context.Context)
  • func (p *SourceWrapper) Start()
• type Stream
  • func InitFromConfig(config config.Configuration) (*Stream, error)
  • func (s *Stream) SetProducer(producer SourceWrapper) error
  • func (s *Stream) SetSinks(sinks []SinkWrapper) error
  • func (s *Stream) Start() error

Types

type ProcessorWrapper

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

ProcessorWrapper wraps processor writer plugin in order to measure performance, build proper configuration and control the context

func NewProcessorWrapper

func NewProcessorWrapper(pluginType processors.ProcessorDriver, config interface{}, appctx *stream_context.Context) ProcessorWrapper

func (*ProcessorWrapper) EvolveSchema

func (p *ProcessorWrapper) EvolveSchema(s *schema.StreamSchemaObj) error

func (*ProcessorWrapper) LoadDriver

func (p *ProcessorWrapper) LoadDriver(driver processors.ProcessorDriver, cfg interface{}) (processors.DataProcessor, error)

func (*ProcessorWrapper) Process

func (p *ProcessorWrapper) Process(msg *message.Message) (*message.Message, error)

func (*ProcessorWrapper) SetStreamContext

func (p *ProcessorWrapper) SetStreamContext(ctx *stream_context.Context)

type SinkWrapper

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

SinkWrapper wraps plan sink writer plugin in order to measure performance, build proper configuration and control the context

func NewSinkWrapper

func NewSinkWrapper(pluginType sinks.SinkDriver, config config.Configuration, appctx *stream_context.Context) SinkWrapper

func (*SinkWrapper) Init

func (p *SinkWrapper) Init() error

func (*SinkWrapper) LoadDriver

func (p *SinkWrapper) LoadDriver(driver sinks.SinkDriver, cfg config.Configuration) sinks.DataSink

func (*SinkWrapper) SetExpectedSchema

func (p *SinkWrapper) SetExpectedSchema(s *schema.StreamSchemaObj)

func (*SinkWrapper) SetStreamContext

func (p *SinkWrapper) SetStreamContext(ctx *stream_context.Context)

func (*SinkWrapper) Write

func (p *SinkWrapper) Write(msg *message.Message) error

type SourceWrapper

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

SourceWrapper wraps source plugin in order to measure performance, build proper configuration and control the context

func NewSourceWrapper

func NewSourceWrapper(pluginType sources.SourceDriver, config config.Configuration) SourceWrapper

func (*SourceWrapper) Events

func (p *SourceWrapper) Events() chan sources.MessageEvent

func (*SourceWrapper) GetPluginConfigs

func (p *SourceWrapper) GetPluginConfigs(driver sources.SourceDriver, fcg *yaml.Node) (any, error)

func (*SourceWrapper) Init

func (p *SourceWrapper) Init(appctx *stream_context.Context) error

func (*SourceWrapper) LoadDriver

func (p *SourceWrapper) LoadDriver(driver sources.SourceDriver, fcg config.Configuration) sources.DataSource

func (*SourceWrapper) SetStreamContext

func (p *SourceWrapper) SetStreamContext(ctx *stream_context.Context)

func (*SourceWrapper) Start

func (p *SourceWrapper) Start()

type Stream

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

func InitFromConfig

func InitFromConfig(config config.Configuration) (*Stream, error)

func (*Stream) SetProducer

func (s *Stream) SetProducer(producer SourceWrapper) error

func (*Stream) SetSinks

func (s *Stream) SetSinks(sinks []SinkWrapper) error

func (*Stream) Start

func (s *Stream) Start() error