forward

Documentation

Overview

Package forward is used for creating a data forwarder for the source vertex.

Index

• type DataForward
  • func NewDataForward(vertexInstance *dfv1.VertexInstance, reader sourcer.SourceReader, ...) (*DataForward, error)
  • func (df *DataForward) ForceStop()
  • func (df *DataForward) IsShuttingDown() (bool, error)
  • func (df *DataForward) Start() <-chan error
  • func (df *DataForward) Stop()
• type Option
  • func WithCallbackUploader(cp *callback.Uploader) Option
  • func WithLogger(l *zap.SugaredLogger) Option
  • func WithReadBatchSize(f int64) Option
  • func WithRetryInterval(f time.Duration) Option
  • func WithTransformer(f applier.SourceTransformApplier) Option
  • func WithTransformerConcurrency(f int) Option
• type Shutdown
  • func (s *Shutdown) String() string

Types

type DataForward

type DataForward struct {
	Shutdown
	// contains filtered or unexported fields
}

DataForward reads data from source and forwards to inter-step buffers

func NewDataForward

func NewDataForward(
	vertexInstance *dfv1.VertexInstance,
	reader sourcer.SourceReader,
	toSteps map[string][]isb.BufferWriter,
	toWhichStepDecider forwarder.ToWhichStepDecider,
	fetchWatermark fetch.SourceFetcher,
	srcWMPublisher publish.SourcePublisher,
	toVertexWmStores map[string]store.WatermarkStore,
	idleManager wmb.IdleManager,
	opts ...Option) (*DataForward, error)

NewDataForward creates a source data forwarder

func (*DataForward) ForceStop

func (df *DataForward) ForceStop()

ForceStop sets up the force shutdown flag.

func (*DataForward) IsShuttingDown

func (df *DataForward) IsShuttingDown() (bool, error)

IsShuttingDown returns whether we can stop processing.

func (*DataForward) Start

func (df *DataForward) Start() <-chan error

Start starts reading from source and forwards to the next buffers. Call `Stop` to stop.

func (*DataForward) Stop

func (df *DataForward) Stop()

Stop stops the processing.

type Option

type Option func(*options) error

func WithCallbackUploader

func WithCallbackUploader(cp *callback.Uploader) Option

WithCallbackUploader sets the callback uploader for the vertex

func WithLogger

func WithLogger(l *zap.SugaredLogger) Option

WithLogger is used to return logger information

func WithReadBatchSize

func WithReadBatchSize(f int64) Option

WithReadBatchSize sets the read batch size

func WithRetryInterval

func WithRetryInterval(f time.Duration) Option

WithRetryInterval sets the retry interval

func WithTransformer

func WithTransformer(f applier.SourceTransformApplier) Option

WithTransformer sets the transformer to be applied

func WithTransformerConcurrency

func WithTransformerConcurrency(f int) Option

WithTransformerConcurrency sets concurrency for source transformer processing

type Shutdown

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

Shutdown tracks and enforces the shutdown activity.

func (*Shutdown) String

func (s *Shutdown) String() string