queue

Oct 8, 2025

Changes in this version

+ var Days = map[string]int

+ var ErrNoJobAvailable = apperror.NewError("no job available")

+ var Months = map[string]int

+ var Presets = map[string]string

+ func IsRetryable(err error) bool

+ type Batch struct

+ Completed int

+ CompletedAt time.Time

+ CreatedAt time.Time

+ Failed int

+ ID string

+ JobIDs []string

+ Metadata map[string]string

+ Name string

+ Pending int

+ Running int

+ Status Status

+ Total int

+ type BatchManager struct

+ func NewBatchManager(manager *Manager) *BatchManager

+ func (bm *BatchManager) CreateBatch(ctx context.Context, name string, jobs []*Job) (*Batch, error)

+ func (bm *BatchManager) DeleteBatch(ctx context.Context, id string) error

+ func (bm *BatchManager) GetBatch(id string) (*Batch, error)

+ func (bm *BatchManager) GetBatches() []*Batch

+ func (bm *BatchManager) UpdateBatchStatus(ctx context.Context, batchID string) error

+ type CronExpression struct

+ Day CronField

+ DayOfWeek CronField

+ Hour CronField

+ Minute CronField

+ Month CronField

+ Second *CronField

+ type CronField struct

+ Max int

+ Min int

+ Values []int

+ type Job struct

+ Attempts int

+ CompletedAt time.Time

+ CreatedAt time.Time

+ Error string

+ ID string

+ MaxAttempts int

+ Metadata map[string]string

+ Payload json.RawMessage

+ Priority Priority

+ Progress float64

+ Results json.RawMessage

+ RetryAt time.Time

+ ScheduleAt time.Time

+ Status Status

+ Timeout time.Duration

+ Type string

+ UpdatedAt time.Time

+ func (j *Job) IsExpired() bool

+ func (j *Job) IsScheduled() bool

+ type JobBuilder struct

+ func NewJob(jobType string) *JobBuilder

+ func (jb *JobBuilder) Build() *Job

+ func (jb *JobBuilder) WithDelay(delay time.Duration) *JobBuilder

+ func (jb *JobBuilder) WithID(id string) *JobBuilder

+ func (jb *JobBuilder) WithJSONPayload(jsonData []byte) *JobBuilder

+ func (jb *JobBuilder) WithMaxAttempts(maxAttempts int) *JobBuilder

+ func (jb *JobBuilder) WithMetadata(key, value string) *JobBuilder

+ func (jb *JobBuilder) WithPayload(payload map[string]interface{}) *JobBuilder

+ func (jb *JobBuilder) WithPriority(priority Priority) *JobBuilder

+ func (jb *JobBuilder) WithScheduleAt(scheduleAt time.Time) *JobBuilder

+ type JobContext struct

+ Job *Job

+ func NewJobContext(_ context.Context, job *Job) *JobContext

+ func (jc *JobContext) GetMetadata(key string) (string, bool)

+ func (jc *JobContext) GetPayload(key string) (interface{}, bool)

+ func (jc *JobContext) GetPayloadBool(key string) (bool, bool)

+ func (jc *JobContext) GetPayloadInt(key string) (int, bool)

+ func (jc *JobContext) GetPayloadString(key string) (string, bool)

+ func (jc *JobContext) ReportProgress(progress float64)

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

+ func LoggingMiddleware(next JobHandler) JobHandler

+ func MetricsMiddleware(next JobHandler) JobHandler

+ func MiddlewareChain(handler JobHandler, middlewares ...Middleware) JobHandler

+ func RecoveryMiddleware(next JobHandler) JobHandler

+ type JobProgress struct

+ JobID string

+ Message string

+ Progress float64

+ type Manager struct

+ func NewManager() *Manager

+ func (m *Manager) Enqueue(ctx context.Context, job *Job) error

+ func (m *Manager) GetJob(ctx context.Context, id string) (*Job, error)

+ func (m *Manager) GetJobs(ctx context.Context, status Status, limit int) ([]*Job, error)

+ func (m *Manager) GetStats() *Stats

+ func (m *Manager) IsRunning() bool

+ func (m *Manager) RegisterHandler(jobType string, handler JobHandler)

+ func (m *Manager) Schedule(ctx context.Context, job *Job) error

+ func (m *Manager) Start(ctx context.Context) error

+ func (m *Manager) Stop() error

+ func (m *Manager) WithQueue(queue Queue) *Manager

+ func (m *Manager) WithRabbitMQ(config RabbitMQConfig) *Manager

+ func (m *Manager) WithRabbitMQFromURL(url string) *Manager

+ func (m *Manager) WithRetryAttempts(attempts int) *Manager

+ func (m *Manager) WithRetryBackoff(backoff float64) *Manager

+ func (m *Manager) WithRetryDelay(delay time.Duration) *Manager

+ func (m *Manager) WithScheduleInterval(interval time.Duration) *Manager

+ func (m *Manager) WithWorkers(workers int) *Manager

+ type MemoryQueue struct

+ func NewMemoryQueue() *MemoryQueue

+ func (mq *MemoryQueue) Close() error

+ func (mq *MemoryQueue) DeleteJob(_ context.Context, id string) error

+ func (mq *MemoryQueue) Dequeue(ctx context.Context, timeout time.Duration) (*Job, error)

+ func (mq *MemoryQueue) Enqueue(_ context.Context, job *Job) error

+ func (mq *MemoryQueue) GetJob(_ context.Context, id string) (*Job, error)

+ func (mq *MemoryQueue) GetJobs(_ context.Context, status Status, limit int) ([]*Job, error)

+ func (mq *MemoryQueue) GetStats(_ context.Context) (*Stats, error)

+ func (mq *MemoryQueue) Schedule(_ context.Context, job *Job) error

+ func (mq *MemoryQueue) UpdateJob(_ context.Context, job *Job) error

+ type Middleware func(JobHandler) JobHandler

+ func TimeoutMiddleware(timeout time.Duration) Middleware

+ type Priority int

+ const PriorityCritical

+ const PriorityHigh

+ const PriorityLow

+ const PriorityNormal

+ func (p Priority) String() string

+ type Queue interface

+ Close func() error

+ DeleteJob func(ctx context.Context, id string) error

+ Dequeue func(ctx context.Context, timeout time.Duration) (*Job, error)

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

+ GetJob func(ctx context.Context, id string) (*Job, error)

+ GetJobs func(ctx context.Context, status Status, limit int) ([]*Job, error)

+ GetStats func(ctx context.Context) (*Stats, error)

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

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

+ type RabbitMQ struct

+ func NewRabbitMQ(config RabbitMQConfig) (*RabbitMQ, error)

+ func NewRabbitMQFromURL(url string) (*RabbitMQ, error)

+ func (rq *RabbitMQ) Close() error

+ func (rq *RabbitMQ) DeleteJob(_ context.Context, id string) error

+ func (rq *RabbitMQ) Dequeue(ctx context.Context, timeout time.Duration) (*Job, error)

+ func (rq *RabbitMQ) Enqueue(ctx context.Context, job *Job) error

+ func (rq *RabbitMQ) GetJob(_ context.Context, id string) (*Job, error)

+ func (rq *RabbitMQ) GetJobs(_ context.Context, status Status, limit int) ([]*Job, error)

+ func (rq *RabbitMQ) GetStats(_ context.Context) (*Stats, error)

+ func (rq *RabbitMQ) IsConnectionOpen() bool

+ func (rq *RabbitMQ) PurgeQueue(_ context.Context) error

+ func (rq *RabbitMQ) Reconnect(config RabbitMQConfig) error

+ func (rq *RabbitMQ) Schedule(ctx context.Context, job *Job) error

+ func (rq *RabbitMQ) UpdateJob(_ context.Context, job *Job) error

+ type RabbitMQConfig struct

+ AutoDelete bool

+ Durable bool

+ ExchangeName string

+ Exclusive bool

+ MaxPriority int

+ NoWait bool

+ QueueName string

+ RoutingKey string

+ URL string

+ type RedisQueue struct

+ func NewRedisQueue(client redis.Cmdable, keyPrefix string) *RedisQueue

+ func (rq *RedisQueue) Close() error

+ func (rq *RedisQueue) DeleteJob(ctx context.Context, id string) error

+ func (rq *RedisQueue) Dequeue(ctx context.Context, timeout time.Duration) (*Job, error)

+ func (rq *RedisQueue) Enqueue(ctx context.Context, job *Job) error

+ func (rq *RedisQueue) GetJob(ctx context.Context, id string) (*Job, error)

+ func (rq *RedisQueue) GetJobs(ctx context.Context, status Status, limit int) ([]*Job, error)

+ func (rq *RedisQueue) GetStats(ctx context.Context) (*Stats, error)

+ func (rq *RedisQueue) MoveScheduledToPending(ctx context.Context) error

+ func (rq *RedisQueue) Schedule(ctx context.Context, job *Job) error

+ func (rq *RedisQueue) UpdateJob(ctx context.Context, job *Job) error

+ type RetryableError struct

+ Err error

+ func NewRetryableError(err error) *RetryableError

+ func (e *RetryableError) Error() string

+ type Stats struct

+ Completed int64

+ DeadLetter int64

+ Failed int64

+ JobsFailed int64

+ JobsProcessed int64

+ JobsRetried int64

+ Pending int64

+ QueueSize int64

+ Retrying int64

+ Running int64

+ Scheduled int64

+ TotalJobs int64

+ WorkersActive int64

+ WorkersBusy int64

+ type Status int

+ const StatusCompleted

+ const StatusDeadLetter

+ const StatusFailed

+ const StatusPending

+ const StatusRetrying

+ const StatusRunning

+ const StatusScheduled

+ func (s Status) String() string

+ type Task struct

+ AllowConcurrent bool

+ ConsecutiveFailures int64

+ CreatedAt time.Time

+ CronSpec string

+ Enabled bool

+ ErrorCount int64

+ Function TaskFunc

+ ID string

+ Interval time.Duration

+ IsRunning bool

+ LastError string

+ LastRun time.Time

+ MaxRetries int

+ Name string

+ NextRun time.Time

+ Quiet bool

+ RetryDelay time.Duration

+ RunCount int64

+ Timeout time.Duration

+ Type TaskType

+ UpdatedAt time.Time

+ type TaskFunc func(ctx context.Context) error

+ type TaskOptions struct

+ Concurrent bool

+ Immediately bool

+ MaxRetries int

+ Quiet bool

+ RetryDelay time.Duration

+ Timeout time.Duration

+ type TaskScheduler struct

+ func NewTaskScheduler() *TaskScheduler

+ func (s *TaskScheduler) DisableTask(name string) error

+ func (s *TaskScheduler) EnableTask(name string) error

+ func (s *TaskScheduler) GetTask(name string) (*Task, error)

+ func (s *TaskScheduler) GetTasks() map[string]*Task

+ func (s *TaskScheduler) IsRunning() bool

+ func (s *TaskScheduler) ParseCronSpec(cronSpec string) (*CronExpression, error)

+ func (s *TaskScheduler) RegisterCronTask(name, cronSpec string, fn TaskFunc) error

+ func (s *TaskScheduler) RegisterCronTaskWithOptions(name, cronSpec string, fn TaskFunc, options TaskOptions) error

+ func (s *TaskScheduler) RegisterIntervalTask(name string, interval time.Duration, fn TaskFunc) error

+ func (s *TaskScheduler) RegisterIntervalTaskWithOptions(name string, interval time.Duration, fn TaskFunc, options TaskOptions) error

+ func (s *TaskScheduler) RegisterOrRescheduleCronTask(name, cronSpec string, fn TaskFunc) error

+ func (s *TaskScheduler) RegisterOrRescheduleCronTaskWithOptions(name, cronSpec string, fn TaskFunc, options TaskOptions) error

+ func (s *TaskScheduler) RegisterOrRescheduleIntervalTask(name string, interval time.Duration, fn TaskFunc) error

+ func (s *TaskScheduler) RegisterOrRescheduleIntervalTaskWithOptions(name string, interval time.Duration, fn TaskFunc, options TaskOptions) error

+ func (s *TaskScheduler) RemoveTask(name string) error

+ func (s *TaskScheduler) RescheduleTaskWithCron(name, cronSpec string) error

+ func (s *TaskScheduler) RescheduleTaskWithInterval(name string, interval time.Duration) error

+ func (s *TaskScheduler) Start(ctx context.Context) error

+ func (s *TaskScheduler) Stop()

+ func (s *TaskScheduler) ValidateCronSpec(spec string) error

+ func (s *TaskScheduler) WithCheckInterval(interval time.Duration) *TaskScheduler

+ func (s *TaskScheduler) WithDefaultTimeout(timeout time.Duration) *TaskScheduler

+ func (s *TaskScheduler) WithRetryDelay(delay time.Duration) *TaskScheduler

+ type TaskType int

+ const TaskTypeCron

+ const TaskTypeInterval

+ func (t TaskType) String() string