operator

type AsyncOperator ¶

type AsyncOperator[T workerpool.TaskMayPanic, R any] struct {
	// contains filtered or unexported fields
}

AsyncOperator process the data in async way.

Eg: The sink of AsyncOperator op1 and the source of op2 use the same channel, Then op2's worker will handle the result from op1.

func NewAsyncOperator[T workerpool.TaskMayPanic, R any](ctx *workerpool.Context, pool *workerpool.WorkerPool[T, R]) *AsyncOperator[T, R]

func NewAsyncOperatorWithTransform[T workerpool.TaskMayPanic, R any](
	ctx *workerpool.Context,
	name string,
	workerNum int,
	transform func(T) R,
) *AsyncOperator[T, R]

func (c *AsyncOperator[T, R]) Close() error

func (c *AsyncOperator[T, R]) GetWorkerPoolSize() int32

func (c *AsyncOperator[T, R]) Open() error

func (c *AsyncOperator[T, R]) SetSink(ch DataChannel[R])

func (c *AsyncOperator[T, R]) SetSource(ch DataChannel[T])

func (*AsyncOperator[T, R]) String() string

func (c *AsyncOperator[T, R]) TuneWorkerPoolSize(workerNum int32, wait bool)

type AsyncPipeline ¶

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

AsyncPipeline wraps a list of Operators. The dataflow is from the first operator to the last operator.

func NewAsyncPipeline(ops ...Operator) *AsyncPipeline

func (p *AsyncPipeline) Close() error

func (p *AsyncPipeline) Execute() error

func (p *AsyncPipeline) GetReaderAndWriter() (operator1, operator2 TunableOperator)

func (p *AsyncPipeline) IsStarted() bool

func (p *AsyncPipeline) String() string

type DataChannel ¶

type DataChannel[T any] interface {
	Channel() chan T
	Finish()
}

DataChannel is a channel that can be used to transfer data between operators.

type Operator ¶

type Operator interface {
	Open() error
	// Close wait task done and close the operator.
	// TODO: the wait part should be separated from the close part.
	Close() error
	String() string
}

Operator is the basic operation unit in the task execution.

type SimpleDataChannel ¶

type SimpleDataChannel[T any] struct {
	// contains filtered or unexported fields
}

SimpleDataChannel is a simple implementation of DataChannel.

func NewSimpleDataChannel[T any](ch chan T) *SimpleDataChannel[T]

func (s *SimpleDataChannel[T]) Channel() chan T

func (s *SimpleDataChannel[T]) Finish()

type SimpleDataSource ¶

type SimpleDataSource[T workerpool.TaskMayPanic] struct {
	// contains filtered or unexported fields
}

SimpleDataSource is a simple operator which use the given input slice as the data source.

func NewSimpleDataSource[T workerpool.TaskMayPanic](
	ctx *workerpool.Context,
	inputs []T,
) *SimpleDataSource[T]

func (s *SimpleDataSource[T]) Close() error

func (s *SimpleDataSource[T]) Open() error

func (s *SimpleDataSource[T]) SetSink(ch DataChannel[T])

func (*SimpleDataSource[T]) String() string

type TunableOperator ¶

type TunableOperator interface {
	TuneWorkerPoolSize(workerNum int32, wait bool)
	GetWorkerPoolSize() int32
}

TunableOperator is the operator which supports modifying pool size.

type WithSink ¶

type WithSink[T any] interface {
	// SetSink sets the sink of the operator.
	// Operator implementations should call the Finish method of the sink when they are done.
	SetSink(channel DataChannel[T])
}

WithSink is an interface that can be used to set the sink of an operator.

type WithSource ¶

type WithSource[T any] interface {
	SetSource(channel DataChannel[T])
}

WithSource is an interface that can be used to set the source of an operator.