queue

Documentation

Overview

Package queue provides a Redis-based job queue using Asynq

Index

• Constants
• type Config
• type DedupeConfig
• type Deduper
  • func NewDeduper(cfg *DedupeConfig) (*Deduper, error)
  • func (d *Deduper) AddIfNotExists(ctx context.Context, key string) bool
  • func (d *Deduper) Clear(ctx context.Context) error
  • func (d *Deduper) Close() error
  • func (d *Deduper) IsDuplicate(ctx context.Context, placeID string) (bool, error)
  • func (d *Deduper) IsDuplicateURL(ctx context.Context, url string) (bool, error)
  • func (d *Deduper) MarkAsSeen(ctx context.Context, placeID string) error
  • func (d *Deduper) Seen(id string) bool
  • func (d *Deduper) Stats(ctx context.Context) (int64, error)
• type JobHandler
• type JobPayload
  • func ParsePayload(data []byte) (*JobPayload, error)
• type Queue
  • func New(cfg *Config) (*Queue, error)
  • func (q *Queue) Close() error
  • func (q *Queue) Enqueue(ctx context.Context, jobID uuid.UUID, priority int) error
  • func (q *Queue) GetQueueStats(ctx context.Context) (map[string]*asynq.QueueInfo, error)
  • func (q *Queue) GetRedisOpt() asynq.RedisConnOpt
• type Worker
  • func NewWorker(cfg *WorkerConfig, handler JobHandler) (*Worker, error)
  • func (w *Worker) Run(ctx context.Context) error
  • func (w *Worker) Shutdown()
• type WorkerConfig

Constants

const (
	// Task types
	TypeJobProcess = "job:process"

	// Queue names for regional affinity
	QueueDefault  = "default"
	QueueHigh     = "high"
	QueueLow      = "low"
	QueueCritical = "critical"
)

Types

type Config

type Config struct {
	RedisURL  string
	RedisAddr string
	Password  string
	DB        int
}

Config holds Redis queue configuration

type DedupeConfig

type DedupeConfig struct {
	RedisURL  string
	RedisAddr string
	Password  string
	DB        int
	Prefix    string
	TTL       time.Duration
}

DedupeConfig holds deduper configuration

type Deduper

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

Deduper provides distributed deduplication using Redis

func NewDeduper

func NewDeduper(cfg *DedupeConfig) (*Deduper, error)

NewDeduper creates a new Redis-based deduplicator

func (*Deduper) AddIfNotExists

func (d *Deduper) AddIfNotExists(ctx context.Context, key string) bool

AddIfNotExists implements the deduper.Deduper interface Returns true if the key was added (not a duplicate), false if already existed

func (*Deduper) Clear

func (d *Deduper) Clear(ctx context.Context) error

Clear removes all dedup keys (use with caution)

func (*Deduper) Close

func (d *Deduper) Close() error

Close closes the Redis connection

func (*Deduper) IsDuplicate

func (d *Deduper) IsDuplicate(ctx context.Context, placeID string) (bool, error)

IsDuplicate checks if a place has already been scraped Returns true if duplicate (already seen), false if new

func (*Deduper) IsDuplicateURL

func (d *Deduper) IsDuplicateURL(ctx context.Context, url string) (bool, error)

IsDuplicateURL checks if a URL has already been processed

func (*Deduper) MarkAsSeen

func (d *Deduper) MarkAsSeen(ctx context.Context, placeID string) error

MarkAsSeen marks a place as seen without checking

func (*Deduper) Seen

func (d *Deduper) Seen(id string) bool

Seen implements the deduper.Deduper interface for compatibility

func (*Deduper) Stats

func (d *Deduper) Stats(ctx context.Context) (int64, error)

Stats returns deduplication statistics

type JobHandler

type JobHandler func(ctx context.Context, payload *JobPayload) error

JobHandler is a function that processes a job

type JobPayload

type JobPayload struct {
	JobID     uuid.UUID `json:"job_id"`
	Priority  int       `json:"priority"`
	CreatedAt time.Time `json:"created_at"`
}

JobPayload is the payload for a job processing task

func ParsePayload

func ParsePayload(data []byte) (*JobPayload, error)

ParsePayload parses a job payload from task data

type Queue

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

Queue is a Redis-based job queue

func New

func New(cfg *Config) (*Queue, error)

New creates a new Queue

func (*Queue) Close

func (q *Queue) Close() error

Close closes the queue client

func (*Queue) Enqueue

func (q *Queue) Enqueue(ctx context.Context, jobID uuid.UUID, priority int) error

Enqueue adds a job to the queue

func (*Queue) GetQueueStats

func (q *Queue) GetQueueStats(ctx context.Context) (map[string]*asynq.QueueInfo, error)

GetQueueStats returns queue statistics

func (*Queue) GetRedisOpt

func (q *Queue) GetRedisOpt() asynq.RedisConnOpt

GetRedisOpt returns the Redis client options for creating a server

type Worker

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

Worker processes jobs from the Redis queue

func NewWorker

func NewWorker(cfg *WorkerConfig, handler JobHandler) (*Worker, error)

NewWorker creates a new queue worker

func (*Worker) Run

func (w *Worker) Run(ctx context.Context) error

Run starts the worker

func (*Worker) Shutdown

func (w *Worker) Shutdown()

Shutdown gracefully shuts down the worker

type WorkerConfig

type WorkerConfig struct {
	RedisURL    string
	RedisAddr   string
	Password    string
	DB          int
	Concurrency int
	Queues      map[string]int // queue name -> priority
}

WorkerConfig holds worker configuration