aboutsummaryrefslogtreecommitdiff
path: root/pkg/worker/list_processor.go
diff options
context:
space:
mode:
Diffstat (limited to 'pkg/worker/list_processor.go')
-rw-r--r--pkg/worker/list_processor.go32
1 files changed, 16 insertions, 16 deletions
diff --git a/pkg/worker/list_processor.go b/pkg/worker/list_processor.go
index 02817c9..ea6b453 100644
--- a/pkg/worker/list_processor.go
+++ b/pkg/worker/list_processor.go
@@ -10,7 +10,7 @@ import (
type (
- // A simple worker to deal with list.
+ // A simple task to deal with list.
ChanProcessor[T any] interface {
Query(context.Context) (<-chan T, error)
Process(context.Context, T) error
@@ -25,62 +25,62 @@ type (
Process(context.Context, T) error
}
- chanProcessorWorker[T any] struct {
+ chanProcessorTask[T any] struct {
chanProcessor ChanProcessor[T]
logrus *logrus.Entry
scheduler *Scheduler
}
- batchProcessorWorker[T any] struct {
+ batchProcessorTask[T any] struct {
batchProcessor ListProcessor[T]
logrus *logrus.Entry
scheduler *Scheduler
}
- serialProcessorWorker[T any] struct {
+ serialProcessorTask[T any] struct {
batchProcessor ListProcessor[T]
logrus *logrus.Entry
scheduler *Scheduler
}
)
-func NewWorkerFromBatchProcessor[T any](
+func NewTaskFromBatchProcessor[T any](
batchProcessor ListProcessor[T],
scheduler *Scheduler,
logrus *logrus.Entry,
-) Worker {
- return &batchProcessorWorker[T]{
+) Task {
+ return &batchProcessorTask[T]{
batchProcessor: batchProcessor,
scheduler: scheduler,
logrus: logrus,
}
}
-func NewWorkerFromSerialProcessor[T any](
+func NewTaskFromSerialProcessor[T any](
batchProcessor ListProcessor[T],
scheduler *Scheduler,
logrus *logrus.Entry,
-) Worker {
- return &serialProcessorWorker[T]{
+) Task {
+ return &serialProcessorTask[T]{
batchProcessor: batchProcessor,
scheduler: scheduler,
logrus: logrus,
}
}
-func NewWorkerFromChanProcessor[T any](
+func NewTaskFromChanProcessor[T any](
chanProcessor ChanProcessor[T],
scheduler *Scheduler,
logrus *logrus.Entry,
-) Worker {
- return &chanProcessorWorker[T]{
+) Task {
+ return &chanProcessorTask[T]{
chanProcessor: chanProcessor,
scheduler: scheduler,
logrus: logrus,
}
}
-func (l *batchProcessorWorker[T]) Start(ctx context.Context) error {
+func (l *batchProcessorTask[T]) Start(ctx context.Context) error {
for {
values, err := l.batchProcessor.Query(ctx)
if err != nil {
@@ -123,7 +123,7 @@ func (l *batchProcessorWorker[T]) Start(ctx context.Context) error {
}
}
-func (l *serialProcessorWorker[T]) Start(ctx context.Context) error {
+func (l *serialProcessorTask[T]) Start(ctx context.Context) error {
for {
values, err := l.batchProcessor.Query(ctx)
if err != nil {
@@ -158,7 +158,7 @@ func (l *serialProcessorWorker[T]) Start(ctx context.Context) error {
}
}
-func (l *chanProcessorWorker[T]) Start(ctx context.Context) error {
+func (l *chanProcessorTask[T]) Start(ctx context.Context) error {
c, err := l.chanProcessor.Query(ctx)
if err != nil {
return err