Documentation
¶
Index ¶
- Constants
- type BackoffConfig
- type BackoffStrategy
- type CommanderFactory
- type Correlation
- type DefaultRetryPolicy
- type Entry
- type Event
- type Hook
- type HookFuncs
- type Option
- func WithCancelKeyExtractor(fn func(*job.ExecutionMessage) string) Option
- func WithCancelPollInterval(interval time.Duration) Option
- func WithCancellationStore(store cancellation.Store) Option
- func WithCommanderFactory(factory CommanderFactory) Option
- func WithConcurrency(concurrency int) Option
- func WithHooks(hooks ...Hook) Option
- func WithIdempotencyStore(store qidempotency.Store, ttl time.Duration) Option
- func WithIdleDelay(delay time.Duration) Option
- func WithLeaseExtensionTTL(ttl time.Duration) Option
- func WithLeaseHeartbeatInterval(interval time.Duration) Option
- func WithLogger(logger job.Logger) Option
- func WithRegistry(registry *Registry) Option
- func WithRetryPolicy(policy RetryPolicy) Option
- func WithShutdownHook(hook ShutdownHook) Option
- func WithTaskCommanderRetries(enabled bool) Option
- type Registry
- type RetryPolicy
- type ShutdownHook
- type Status
- type Worker
- func (w *Worker) Pause() error
- func (w *Worker) Register(task job.Task) error
- func (w *Worker) RegisterAll(tasks []job.Task) error
- func (w *Worker) RegisteredTasks() []job.Task
- func (w *Worker) Resume() error
- func (w *Worker) Start(ctx context.Context) error
- func (w *Worker) Status() Status
- func (w *Worker) Stop(ctx context.Context) error
Constants ¶
const ( // WorkerStatusRunning means the worker is actively consuming deliveries. WorkerStatusRunning = "running" // WorkerStatusPaused means consumption is paused but worker goroutines remain active. WorkerStatusPaused = "paused" // WorkerStatusStopped means the worker has been stopped. WorkerStatusStopped = "stopped" )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BackoffConfig ¶
type BackoffConfig struct {
Strategy BackoffStrategy
Interval time.Duration
MaxInterval time.Duration
Jitter bool
}
BackoffConfig configures retry delays.
type BackoffStrategy ¶
type BackoffStrategy string
BackoffStrategy controls the retry delay profile.
const ( BackoffNone BackoffStrategy = "none" BackoffFixed BackoffStrategy = "fixed" BackoffExponential BackoffStrategy = "exponential" )
type CommanderFactory ¶
type CommanderFactory func(job.Task) *job.TaskCommander
CommanderFactory builds a TaskCommander for a task.
type Correlation ¶ added in v0.15.0
type Correlation struct {
MachineID string
EntityID string
ExecutionID string
ExpectedState string
ExpectedVersion int64
ResumeEvent string
}
Correlation captures canonical FSM metadata for observability hooks.
type DefaultRetryPolicy ¶
type DefaultRetryPolicy struct {
MaxAttempts int
Backoff BackoffConfig
}
DefaultRetryPolicy provides simple max-attempts with backoff scheduling. MaxAttempts <= 0 means no retries (dead-letter on first failure).
func (DefaultRetryPolicy) Decide ¶
func (p DefaultRetryPolicy) Decide(attempt int, err error) queue.NackOptions
Decide returns nack options based on attempts and policy.
type Entry ¶
type Entry struct {
Task job.Task
Commander *job.TaskCommander
}
Entry stores a registered task and its commander.
type Event ¶
type Event struct {
Delivery queue.Delivery
Message *job.ExecutionMessage
Task job.Task
Correlation Correlation
Attempt int
Delay time.Duration
Err error
StartedAt time.Time
Duration time.Duration
}
Event captures worker lifecycle details for hooks.
type Hook ¶
type Hook interface {
OnStart(ctx context.Context, event Event)
OnSuccess(ctx context.Context, event Event)
OnFailure(ctx context.Context, event Event)
OnRetry(ctx context.Context, event Event)
}
Hook exposes lifecycle callbacks for worker execution.
type HookFuncs ¶
type HookFuncs struct {
OnStartFunc func(context.Context, Event)
OnSuccessFunc func(context.Context, Event)
OnFailureFunc func(context.Context, Event)
OnRetryFunc func(context.Context, Event)
}
HookFuncs provides a function-based hook implementation.
type Option ¶
type Option func(*Worker)
Option configures a Worker instance.
func WithCancelKeyExtractor ¶
func WithCancelKeyExtractor(fn func(*job.ExecutionMessage) string) Option
WithCancelKeyExtractor sets the key extractor for cancellation lookups.
func WithCancelPollInterval ¶
WithCancelPollInterval sets the polling interval for cancellation checks.
func WithCancellationStore ¶
func WithCancellationStore(store cancellation.Store) Option
WithCancellationStore configures a distributed cancellation store.
func WithCommanderFactory ¶
func WithCommanderFactory(factory CommanderFactory) Option
WithCommanderFactory overrides the TaskCommander construction.
func WithConcurrency ¶
WithConcurrency sets the number of worker goroutines.
func WithIdempotencyStore ¶ added in v0.15.0
func WithIdempotencyStore(store qidempotency.Store, ttl time.Duration) Option
WithIdempotencyStore configures distributed deduplication across workers.
func WithIdleDelay ¶
WithIdleDelay controls the delay between dequeue attempts when idle.
func WithLeaseExtensionTTL ¶ added in v0.15.0
WithLeaseExtensionTTL sets the lease extension duration used by heartbeats.
func WithLeaseHeartbeatInterval ¶ added in v0.15.0
WithLeaseHeartbeatInterval sets the cadence for lease renewal heartbeats.
func WithLogger ¶
WithLogger injects a logger for worker events.
func WithRegistry ¶
WithRegistry supplies a task registry.
func WithRetryPolicy ¶
func WithRetryPolicy(policy RetryPolicy) Option
WithRetryPolicy sets the retry policy for nacks.
func WithShutdownHook ¶
func WithShutdownHook(hook ShutdownHook) Option
WithShutdownHook registers a shutdown hook.
func WithTaskCommanderRetries ¶
WithTaskCommanderRetries enables or disables TaskCommander retries.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry stores tasks by job_id and validates registrations.
type RetryPolicy ¶
type RetryPolicy interface {
Decide(attempt int, err error) queue.NackOptions
}
RetryPolicy decides how the worker should nack a failed delivery.
type ShutdownHook ¶
ShutdownHook runs during worker shutdown.
type Status ¶ added in v0.15.0
type Status struct {
Status string
Running bool
Paused bool
Concurrency int
Registered int
StartedAt time.Time
StoppedAt time.Time
}
Status reports worker control/runtime state for orchestration surfaces.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker consumes queue deliveries and dispatches tasks.
func (*Worker) Pause ¶ added in v0.15.0
Pause temporarily halts delivery consumption without stopping worker goroutines.
func (*Worker) RegisterAll ¶
RegisterAll adds multiple tasks.
func (*Worker) RegisteredTasks ¶
RegisteredTasks returns the current task list.
Source Files