worker

Jan 24, 2026

Changes in this version

+ type TaskHooks struct

+ OnFinish func(task *Task, status TaskStatus, result any, err error)

+ OnQueued func(task *Task)

+ OnRetry func(task *Task, delay time.Duration, attempt int)

+ OnStart func(task *Task)

type TaskManager

+ func (tm *TaskManager) SetHooks(hooks TaskHooks)

Jan 23, 2026

Changes in this version

+ const ErrMsgContextDone

type MetricsSnapshot

+ QueueDepth int

+ ResultsDropped int64

+ Retried int64

+ Running int64

+ TaskLatencyCount int64

+ TaskLatencyMax time.Duration

+ TaskLatencyTotal time.Duration

+ type RetentionPolicy struct

+ CleanupInterval time.Duration

+ MaxEntries int

+ TTL time.Duration

type Task

+ func (task *Task) CancelledAt() time.Time

+ func (task *Task) CompletedAt() time.Time

+ func (task *Task) StartedAt() time.Time

type TaskManager

+ func (tm *TaskManager) SetRetentionPolicy(policy RetentionPolicy)

+ func (tm *TaskManager) StopGraceful(ctx context.Context) error

+ func (tm *TaskManager) StopNow()

+ func (tm *TaskManager) SubscribeResults(buffer int) (<-chan Result, func())

Aug 18, 2025

Changes in this version

+ type GRPCServer struct

+ func NewGRPCServer(svc Service, handlers map[string]HandlerSpec) *GRPCServer

+ func (s *GRPCServer) CancelTask(ctx context.Context, req *workerpb.CancelTaskRequest) (*workerpb.CancelTaskResponse, error)

+ func (s *GRPCServer) GetTask(ctx context.Context, req *workerpb.GetTaskRequest) (*workerpb.GetTaskResponse, error)

+ func (s *GRPCServer) RegisterTasks(ctx context.Context, req *workerpb.RegisterTasksRequest) (*workerpb.RegisterTasksResponse, error)

+ func (s *GRPCServer) StreamResults(req *workerpb.StreamResultsRequest, ...) error

+ type HandlerSpec struct

+ Fn func(ctx context.Context, payload protoreflect.ProtoMessage) (any, error)

+ Make func() protoreflect.ProtoMessage

Aug 16, 2025

Changes in this version

+ type MetricsSnapshot struct

+ Cancelled int64

+ Completed int64

+ Failed int64

+ Scheduled int64

type TaskManager

+ ExecuteTaskErrors *ewrap.ErrorGroup

+ RegisterTaskErrors *ewrap.ErrorGroup

+ func (tm *TaskManager) GetMetrics() MetricsSnapshot

+ func (tm *TaskManager) SetMaxWorkers(maxWorkers int)

May 19, 2023

Changes in this version

type Task

+ Result atomic.Value

May 19, 2023

Changes in this version

+ const Completed

+ const Invalid

+ var ErrInvalidTaskContext = errors.New("invalid task context")

type Service

+ GetCancelledTasks func() <-chan Task

type Task

+ Execute TaskFunc

+ func NewTask(fn TaskFunc, ctx context.Context) (*Task, error)

type TaskManager

+ func (tm *TaskManager) GetCancelledTasks() <-chan Task

+ func (tm *TaskManager) IsEmpty() bool

May 17, 2023

Changes in this version

type Service

+ StreamResults func() <-chan Result

type Task

+ func (task *Task) ShouldSchedule() error

type TaskManager

+ func NewTaskManagerWithDefaults() *TaskManager

+ func (tm *TaskManager) StreamResults() <-chan Result

Mar 6, 2023

Changes in this version

+ const DefaultMaxRetries

+ const DefaultMaxTasks

+ const DefaultRetryDelay

+ const DefaultTasksPerSecond

+ const DefaultTimeout

+ const Failed

+ const Queued

+ const Running

+ var ErrTaskAlreadyStarted = errors.New("task already started")

+ var ErrTaskCancelled = errors.New("task cancelled")

+ var ErrTaskCompleted = errors.New("task completed")

+ var ErrTaskTimeout = errors.New("task timeout")

+ type Result struct

+ Error error

+ Result interface{}

+ Task *Task

+ func (r *Result) String() string

type Service

+ CancelAllAndWait func()

+ Close func()

+ GetActiveTasks func() int

+ GetCancelled func() <-chan Task

+ RegisterTasks func(ctx context.Context, tasks ...Task)

+ StartWorkers func()

+ Wait func(timeout time.Duration)

type Task

+ CancelFunc context.CancelFunc

+ Description string

+ Name string

+ Retries int

+ RetryDelay time.Duration

+ Status TaskStatus

+ func (task *Task) CancelledChan() <-chan struct{}

+ func (task *Task) ShouldExecute() error

+ func (task *Task) WaitCancelled()

type TaskManager

+ Cancelled chan Task

+ MaxRetries int

+ MaxTasks int

+ MaxWorkers int

+ RetryDelay time.Duration

+ Tasks chan Task

+ Timeout time.Duration

+ func (tm *TaskManager) CancelAllAndWait()

+ func (tm *TaskManager) Close()

+ func (tm *TaskManager) GetActiveTasks() int

+ func (tm *TaskManager) GetCancelled() <-chan Task

+ func (tm *TaskManager) RegisterTasks(ctx context.Context, tasks ...Task)

+ func (tm *TaskManager) StartWorkers()

+ func (tm *TaskManager) Wait(timeout time.Duration)

+ type TaskStatus uint8

+ func (ts TaskStatus) String() string

Jan 29, 2023

Changes in this version

+ const Cancelled

+ const ContextDeadlineReached

+ const RateLimited

+ var ErrInvalidTaskFunc = errors.New("invalid task function")

+ var ErrInvalidTaskID = errors.New("invalid task id")

+ var ErrTaskNotFound = errors.New("task not found")

+ type CancelReason uint8

type Service

+ ExecuteTask func(id uuid.UUID) (interface{}, error)

type Task

+ CancelReason CancelReason

+ Error atomic.Value

+ func (t *Task) IsValid() (err error)

+ type TaskFunc func() interface

type TaskManager

+ func (tm *TaskManager) ExecuteTask(id uuid.UUID) (interface{}, error)

Jan 28, 2023

Changes in this version

+ type Middleware func(Service) Service

+ type Service interface

+ CancelAll func()

+ CancelTask func(id uuid.UUID)

+ GetResults func() <-chan interface{}

+ GetTask func(id uuid.UUID) (task Task, ok bool)

+ GetTasks func() []Task

+ RegisterTask func(tasks ...Task)

+ Start func(numWorkers int)

+ Stop func()

+ func NewTaskManager(maxTasks int, limit float64) Service

+ func RegisterMiddleware(svc Service, mw ...Middleware) Service

+ type Task struct

+ Cancel context.CancelFunc

+ Cancelled atomic.Int64

+ Completed atomic.Int64

+ Ctx context.Context

+ Fn func() interface{}

+ ID uuid.UUID

+ Priority int

+ Started atomic.Int64

+ type TaskManager struct

+ Registry sync.Map

+ Results chan interface{}

+ func (tm *TaskManager) CancelAll()

+ func (tm *TaskManager) CancelTask(id uuid.UUID)

+ func (tm *TaskManager) GetResults() <-chan interface{}

+ func (tm *TaskManager) GetTask(id uuid.UUID) (task Task, ok bool)

+ func (tm *TaskManager) GetTasks() []Task

+ func (tm *TaskManager) RegisterTask(tasks ...Task)

+ func (tm *TaskManager) Start(numWorkers int)

+ func (tm *TaskManager) Stop()