taskqueue

package module

Go to main page

Versions in this module

v1

v1.1.2

Feb 3, 2025

v1.1.1

Feb 3, 2025

v1.1.0

Jan 31, 2025

Changes in this version

+ const MetricJobFailedCount

+ type ErrSkipRetry struct

+ Err error

+ SkipReason string

+ func (e ErrSkipRetry) Error() string

+ func (e ErrSkipRetry) Unwrap() error

type HeartbeatData

+ CPUUsage float64

+ MemoryUsage float64

type NackOptions

+ ShouldRetry bool

v1.0.0

Jan 30, 2025

Changes in this version

+ const DefaultNameSpace

+ const MetricDeadQueueSize

+ const MetricJobProcessedCount

+ const MetricPendingQueueSize

+ var ErrInvalidJobStatus = errors.New("invalid job status")

+ var ErrJobNotFound = errors.New("job not found")

+ type AckOptions struct

+ QueueName string

+ type Acker interface

+ Ack func(ctx context.Context, job *Job, opts *AckOptions) error

+ Nack func(ctx context.Context, job *Job, opts *NackOptions) error

+ type Counter interface

+ IncrementCounter func(ctx context.Context, m Metric, count int, ts time.Time) error

+ QueryRangeCounterValues func(ctx context.Context, m Metric, start, end time.Time) (MetricRangeValue, error)

+ type DequeueFunc func(ctx context.Context, opts *DequeueOptions, count int) ([]*Job, error)

+ func (f DequeueFunc) Dequeue(ctx context.Context, opts *DequeueOptions, count int) ([]*Job, error)

+ type DequeueOptions struct

+ JobTimeout time.Duration

+ QueueName string

+ type Dequeuer interface

+ Dequeue func(ctx context.Context, opts *DequeueOptions, count int) ([]*Job, error)

+ type EnqueueOptions struct

+ Delay time.Duration

+ QueueName string

+ type Enqueuer interface

+ Enqueue func(ctx context.Context, job *Job, opts *EnqueueOptions) error

+ type Gauge interface

+ GaugeValue func(ctx context.Context, m Metric) (MetricValue, error)

+ QueryRangeGaugeValues func(ctx context.Context, m Metric, start, end time.Time) (MetricRangeValue, error)

+ RecordGauge func(ctx context.Context, m Metric, value float64, ts time.Time) error

+ type Handler interface

+ Handle func(ctx context.Context, job *Job) error

+ type HandlerFunc func(context.Context, *Job) error

+ func (h HandlerFunc) Handle(ctx context.Context, job *Job) error

+ type HeartBeater interface

+ LastHeartbeats func(ctx context.Context) ([]HeartbeatData, error)

+ RemoveHeartbeat func(ctx context.Context, workerID string) error

+ SendHeartbeat func(ctx context.Context, data HeartbeatData) error

+ type HeartbeatData struct

+ HeartbeatAt time.Time

+ PID int

+ Queues []HeartbeatQueueData

+ StartedAt time.Time

+ WorkerID string

+ type HeartbeatQueueData struct

+ Concurrency int

+ MaxAttempts int

+ Name string

+ Timeout time.Duration

+ type Job struct

+ Attempts int

+ CreatedAt time.Time

+ FailureReason string

+ ID string

+ Payload []byte

+ ProcessedBy string

+ QueueName string

+ StartedAt time.Time

+ Status JobStatus

+ UpdatedAt time.Time

+ func NewJob() *Job

+ func (j *Job) JSONMarshalPayload(v any) (err error)

+ func (j *Job) JSONUnMarshalPayload(v any) error

+ type JobOption func(*JobOptions)

+ func WithBackoffFunc(f func(attempts int) time.Duration) JobOption

+ func WithConcurrency(concurrency int) JobOption

+ func WithIdleWaitTime(idleWaitTime time.Duration) JobOption

+ func WithMaxAttempts(maxAttempts int) JobOption

+ func WithTimeout(timeout time.Duration) JobOption

+ type JobOptions struct

+ BackoffFunc func(attempts int) time.Duration

+ Concurrency int

+ IdleWaitTime time.Duration

+ MaxAttempts int

+ Timeout time.Duration

+ type JobStatus int8

+ const JobStatusActive

+ const JobStatusCompleted

+ const JobStatusDead

+ const JobStatusFailed

+ const JobStatusWaiting

+ func ParseJobStatus(text string) (JobStatus, error)

+ func (j JobStatus) String() string

+ type Logger interface

+ Debug func(format string, args ...interface{})

+ Error func(format string, args ...interface{})

+ Info func(format string, args ...interface{})

+ type Metric struct

+ Labels map[string]string

+ Name string

+ type MetricRangeValue struct

+ Metric Metric

+ Values []MetricValue

+ type MetricValue struct

+ TimeStamp time.Time

+ Value float64

+ type Metrics interface

+ type NackOptions struct

+ MaxAttemptsExceeded bool

+ QueueName string

+ RetryAfter time.Duration

+ type Pagination struct

+ Page int

+ Rows int

+ type Queue interface

+ type QueueDetails struct

+ JobCount int

+ Jobs []*Job

+ Name string

+ NameSpace string

+ Pagination Pagination

+ Status QueueStatus

+ type QueueError int

+ const ErrQueueEmpty

+ const ErrQueueNotFound

+ const ErrUnknown

+ func (err QueueError) Error() string

+ type QueueInfo struct

+ JobCount int

+ Name string

+ NameSpace string

+ Status QueueStatus

+ type QueueManager interface

+ DeleteJobFromDeadQueue func(ctx context.Context, queueName string, jobID string) error

+ ListDeadQueues func(ctx context.Context) ([]*QueueInfo, error)

+ ListPendingQueues func(ctx context.Context) ([]*QueueInfo, error)

+ PageDeadQueue func(ctx context.Context, queueName string, p Pagination) (*QueueDetails, error)

+ PagePendingQueue func(ctx context.Context, queueName string, p Pagination) (*QueueDetails, error)

+ PausePendingQueue func(ctx context.Context, queueName string) error

+ ResumePendingQueue func(ctx context.Context, queueName string) error

+ type QueueStatus int

+ const QueueStatusPaused

+ const QueueStatusRunning

+ const QueueStatusUnknown

+ func (s QueueStatus) String() string

+ type Worker struct

+ func NewWorker(opts *WorkerOptions) *Worker

+ func (w *Worker) RegisterHandler(queueName string, h Handler, opts ...JobOption)

+ func (w *Worker) Start(ctx context.Context)

+ func (w *Worker) Stop()

+ type WorkerOptions struct

+ ErrorHandler func(err error)

+ HeartBeater HeartBeater

+ ID string

+ Logger Logger

+ MetricsBackend Metrics

+ Queue Queue

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL