queue

Documentation

Index

• type Config
  • func DefaultConfig() *Config
• type Job
• type JobHandler
• type JobPriority
  • func (p JobPriority) String() string
• type JobResult
• type JobStatus
• type Queue
  • func New(cache *cache.Cache, config *Config, handler JobHandler, logger *logrus.Logger) *Queue
  • func (q *Queue) CancelJob(ctx context.Context, jobID string) error
  • func (q *Queue) Enqueue(ctx context.Context, job *Job) error
  • func (q *Queue) GetJob(ctx context.Context, jobID string) (*Job, error)
  • func (q *Queue) GetQueueStats(ctx context.Context) (map[string]interface{}, error)
  • func (q *Queue) SetHandler(handler JobHandler)
  • func (q *Queue) Start(ctx context.Context) error
  • func (q *Queue) Stop() error
  • func (q *Queue) UpdateJob(ctx context.Context, job *Job) error

Types

type Config

type Config struct {
	Workers         int           `yaml:"workers"`
	MaxRetries      int           `yaml:"max_retries"`
	RetryBackoff    time.Duration `yaml:"retry_backoff"`
	JobTimeout      time.Duration `yaml:"job_timeout"`
	PollingInterval time.Duration `yaml:"polling_interval"`
	RetentionPeriod time.Duration `yaml:"retention_period"`
}

Config represents queue configuration

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig returns default queue configuration

type Job

type Job struct {
	ID           string                 `json:"id"`
	FunctionID   string                 `json:"function_id"`
	FunctionName string                 `json:"function_name"`
	Payload      map[string]interface{} `json:"payload"`
	Priority     JobPriority            `json:"priority"`
	Status       JobStatus              `json:"status"`
	Attempts     int                    `json:"attempts"`
	MaxRetries   int                    `json:"max_retries"`
	Timeout      time.Duration          `json:"timeout"`
	CreatedAt    time.Time              `json:"created_at"`
	StartedAt    *time.Time             `json:"started_at,omitempty"`
	CompletedAt  *time.Time             `json:"completed_at,omitempty"`
	Result       *JobResult             `json:"result,omitempty"`
	Error        string                 `json:"error,omitempty"`
	NextRetryAt  *time.Time             `json:"next_retry_at,omitempty"`
}

Job represents a queued execution job

type JobHandler

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

JobHandler processes jobs from the queue

type JobPriority

type JobPriority int

JobPriority represents the priority of a job

const (
	PriorityLow    JobPriority = 0
	PriorityNormal JobPriority = 1
	PriorityHigh   JobPriority = 2
	PriorityUrgent JobPriority = 3
)

func (JobPriority) String

func (p JobPriority) String() string

Priority string representation

type JobResult

type JobResult struct {
	Output      string        `json:"output"`
	Duration    time.Duration `json:"duration"`
	ContainerID string        `json:"container_id"`
	ExitCode    int           `json:"exit_code"`
}

JobResult represents the result of a job execution

type JobStatus

type JobStatus string

JobStatus represents the status of a queued job

const (
	JobStatusPending   JobStatus = "pending"
	JobStatusRunning   JobStatus = "running"
	JobStatusCompleted JobStatus = "completed"
	JobStatusFailed    JobStatus = "failed"
	JobStatusRetrying  JobStatus = "retrying"
	JobStatusCancelled JobStatus = "canceled"
)

type Queue

type Queue struct {
	// contains filtered or unexported fields
}

Queue represents an async job queue

func New

func New(cache *cache.Cache, config *Config, handler JobHandler, logger *logrus.Logger) *Queue

New creates a new job queue

func (*Queue) CancelJob

func (q *Queue) CancelJob(ctx context.Context, jobID string) error

CancelJob cancels a pending or running job

func (*Queue) Enqueue

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

Enqueue adds a job to the queue

func (*Queue) GetJob

func (q *Queue) GetJob(ctx context.Context, jobID string) (*Job, error)

GetJob retrieves a job by ID

func (*Queue) GetQueueStats

func (q *Queue) GetQueueStats(ctx context.Context) (map[string]interface{}, error)

GetQueueStats returns queue statistics

func (*Queue) SetHandler

func (q *Queue) SetHandler(handler JobHandler)

SetHandler updates the job handler

func (*Queue) Start

func (q *Queue) Start(ctx context.Context) error

Start starts the queue workers

func (*Queue) Stop

func (q *Queue) Stop() error

Stop stops the queue workers

func (*Queue) UpdateJob

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

UpdateJob updates a job in the queue