forward

Documentation

Index

• func DefaultOptions() *options
• type DataForward
  • func NewDataForward(vertexInstance *dfv1.VertexInstance, fromStep isb.BufferReader, ...) (*DataForward, error)
  • func (df *DataForward) ForceStop()
  • func (df *DataForward) IsShuttingDown() (bool, error)
  • func (df *DataForward) Start() <-chan struct{}
  • func (df *DataForward) Stop()
• type Option
  • func WithFbSinkWriter(sinkWriter sinker.SinkWriter) Option
  • func WithLogger(l *zap.SugaredLogger) Option
  • func WithReadBatchSize(f int64) Option
  • func WithRetryInterval(f time.Duration) Option
  • func WithSinkConcurrency(f int) Option
• type Shutdown
  • func (s *Shutdown) String() string

Functions

func DefaultOptions

func DefaultOptions() *options

Types

type DataForward

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

DataForward forwards the data from previous vertex to sinker.

func NewDataForward

func NewDataForward(
	vertexInstance *dfv1.VertexInstance,
	fromStep isb.BufferReader,
	sinkWriter sinker.SinkWriter,
	fetchWatermark fetch.Fetcher,
	publishWatermark publish.Publisher,
	idleManager wmb.IdleManager,
	opts ...Option) (*DataForward, error)

NewDataForward creates a sink 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 struct{}

Start starts reading the buffer and forwards to sinker. Call `Stop` to stop.

func (*DataForward) Stop

func (df *DataForward) Stop()

Stop stops the processing.

type Option

type Option func(*options) error

func WithFbSinkWriter

func WithFbSinkWriter(sinkWriter sinker.SinkWriter) Option

WithFbSinkWriter sets the fallback sink writer

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 WithSinkConcurrency

func WithSinkConcurrency(f int) Option

WithSinkConcurrency sets concurrency for 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