processing

Documentation

Index

• func ProcessStream(stream kodex.Stream, timeout time.Duration) error
• type ConfigContext
• type Executor
• type InternalReader
• type LocalDestinationWorker
  • func MakeLocalDestinationWorker(pool chan chan kodex.Payload, writer kodex.Writer, executor Executor) (*LocalDestinationWorker, error)
  • func (w *LocalDestinationWorker) ProcessPayload(payload kodex.Payload) error
  • func (w *LocalDestinationWorker) Start()
  • func (w *LocalDestinationWorker) Stop()
• type LocalDestinationWriter
  • func MakeLocalDestinationWriter(maxDestinationWorkers int, id []byte) *LocalDestinationWriter
  • func (d *LocalDestinationWriter) ID() []byte
  • func (d *LocalDestinationWriter) Start(supervisor Supervisor, processable kodex.Processable) error
  • func (d *LocalDestinationWriter) Stop(graceful bool) error
  • func (d *LocalDestinationWriter) Stopped() bool
• type LocalSourceReader
  • func MakeLocalSourceReader(maxSourceWorkers int, id []byte) *LocalSourceReader
  • func (d *LocalSourceReader) ID() []byte
  • func (d *LocalSourceReader) SourceMap() kodex.SourceMap
  • func (d *LocalSourceReader) Start(supervisor Supervisor, processable kodex.Processable) error
  • func (d *LocalSourceReader) Stop(graceful bool) error
  • func (d *LocalSourceReader) Stopped() bool
• type LocalSourceWorker
  • func MakeLocalSourceWorker(pool chan chan kodex.Payload, streams []kodex.Stream, executor Executor) (*LocalSourceWorker, error)
  • func (w *LocalSourceWorker) ProcessPayload(payload kodex.Payload) error
  • func (w *LocalSourceWorker) Start()
  • func (w *LocalSourceWorker) Stop()
• type LocalStreamExecutor
  • func MakeLocalStreamExecutor(maxStreamWorkers int, id []byte) *LocalStreamExecutor
  • func (d *LocalStreamExecutor) ID() []byte
  • func (d *LocalStreamExecutor) Start(supervisor Supervisor, processable kodex.Processable) error
  • func (d *LocalStreamExecutor) Stop(graceful bool) error
  • func (d *LocalStreamExecutor) Stopped() bool
  • func (d *LocalStreamExecutor) Stream() kodex.Stream
• type LocalStreamWorker
  • func MakeLocalStreamWorker(pool chan chan kodex.Payload, contexts []*ConfigContext, ...) (*LocalStreamWorker, error)
  • func (w *LocalStreamWorker) ProcessPayload(payload kodex.Payload) error
  • func (w *LocalStreamWorker) Start()
  • func (w *LocalStreamWorker) Stop()
• type Supervisor

Functions

func ProcessStream

func ProcessStream(stream kodex.Stream, timeout time.Duration) error

Types

type ConfigContext

type ConfigContext struct {
	Config       kodex.Config
	Processor    *kodex.Processor
	Destinations map[string][]kodex.DestinationMap
}

type Executor

type Executor interface {
	Start(Supervisor, kodex.Processable) error
	Stop(graceful bool) error
	Stopped() bool
	ID() []byte
}

type InternalReader

type InternalReader interface {
	Read() (kodex.Payload, error)
	ID() []byte
}

An internal reader reads items for a stream from an internal source

type LocalDestinationWorker

type LocalDestinationWorker struct {
	ItemsProcessed int
	// contains filtered or unexported fields
}

func MakeLocalDestinationWorker

func MakeLocalDestinationWorker(pool chan chan kodex.Payload,
	writer kodex.Writer,
	executor Executor) (*LocalDestinationWorker, error)

func (*LocalDestinationWorker) ProcessPayload

func (w *LocalDestinationWorker) ProcessPayload(payload kodex.Payload) error

func (*LocalDestinationWorker) Start

func (w *LocalDestinationWorker) Start()

func (*LocalDestinationWorker) Stop

func (w *LocalDestinationWorker) Stop()

type LocalDestinationWriter

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

func MakeLocalDestinationWriter

func MakeLocalDestinationWriter(maxDestinationWorkers int,
	id []byte) *LocalDestinationWriter

func (*LocalDestinationWriter) ID

func (d *LocalDestinationWriter) ID() []byte

func (*LocalDestinationWriter) Start

func (d *LocalDestinationWriter) Start(supervisor Supervisor, processable kodex.Processable) error

func (*LocalDestinationWriter) Stop

func (d *LocalDestinationWriter) Stop(graceful bool) error

func (*LocalDestinationWriter) Stopped

func (d *LocalDestinationWriter) Stopped() bool

type LocalSourceReader

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

func MakeLocalSourceReader

func MakeLocalSourceReader(maxSourceWorkers int,
	id []byte) *LocalSourceReader

func (*LocalSourceReader) ID

func (d *LocalSourceReader) ID() []byte

func (*LocalSourceReader) SourceMap

func (d *LocalSourceReader) SourceMap() kodex.SourceMap

func (*LocalSourceReader) Start

func (d *LocalSourceReader) Start(supervisor Supervisor, processable kodex.Processable) error

func (*LocalSourceReader) Stop

func (d *LocalSourceReader) Stop(graceful bool) error

func (*LocalSourceReader) Stopped

func (d *LocalSourceReader) Stopped() bool

type LocalSourceWorker

type LocalSourceWorker struct {
	ItemsProcessed int
	// contains filtered or unexported fields
}

func MakeLocalSourceWorker

func MakeLocalSourceWorker(pool chan chan kodex.Payload,
	streams []kodex.Stream,
	executor Executor) (*LocalSourceWorker, error)

func (*LocalSourceWorker) ProcessPayload

func (w *LocalSourceWorker) ProcessPayload(payload kodex.Payload) error

func (*LocalSourceWorker) Start

func (w *LocalSourceWorker) Start()

func (*LocalSourceWorker) Stop

func (w *LocalSourceWorker) Stop()

type LocalStreamExecutor

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

func MakeLocalStreamExecutor

func MakeLocalStreamExecutor(maxStreamWorkers int,
	id []byte) *LocalStreamExecutor

func (*LocalStreamExecutor) ID

func (d *LocalStreamExecutor) ID() []byte

func (*LocalStreamExecutor) Start

func (d *LocalStreamExecutor) Start(supervisor Supervisor, processable kodex.Processable) error

func (*LocalStreamExecutor) Stop

func (d *LocalStreamExecutor) Stop(graceful bool) error

func (*LocalStreamExecutor) Stopped

func (d *LocalStreamExecutor) Stopped() bool

func (*LocalStreamExecutor) Stream

func (d *LocalStreamExecutor) Stream() kodex.Stream

type LocalStreamWorker

type LocalStreamWorker struct {
	ItemsProcessed int
	// contains filtered or unexported fields
}

func MakeLocalStreamWorker

func MakeLocalStreamWorker(pool chan chan kodex.Payload,
	contexts []*ConfigContext,
	acknowledgeFailed bool,
	executor Executor) (*LocalStreamWorker, error)

func (*LocalStreamWorker) ProcessPayload

func (w *LocalStreamWorker) ProcessPayload(payload kodex.Payload) error

func (*LocalStreamWorker) Start

func (w *LocalStreamWorker) Start()

func (*LocalStreamWorker) Stop

func (w *LocalStreamWorker) Stop()

type Supervisor

type Supervisor interface {
	ExecutorStopped(Executor, kodex.Processable)
}