Documentation
¶
Overview ¶
Package queue provides foxhound.Queue implementations. The memory package offers an in-process, priority-ordered queue suitable for single-node scraping runs.
Index ¶
- Variables
- func NewMemoryQueue() foxhound.Queue
- type RedisQueue
- type ReliableConfig
- type ReliableQueue
- func (rq *ReliableQueue) Ack(job *foxhound.Job)
- func (rq *ReliableQueue) Close() error
- func (rq *ReliableQueue) DLQ() []*foxhound.Job
- func (rq *ReliableQueue) DLQLen() int
- func (rq *ReliableQueue) InFlightLen() int
- func (rq *ReliableQueue) Len() int
- func (rq *ReliableQueue) Nack(ctx context.Context, job *foxhound.Job)
- func (rq *ReliableQueue) Pop(ctx context.Context) (*foxhound.Job, error)
- func (rq *ReliableQueue) Push(ctx context.Context, job *foxhound.Job) error
- func (rq *ReliableQueue) RetryDLQ(ctx context.Context) (int, error)
- type SQLiteQueue
Constants ¶
This section is empty.
Variables ¶
var ErrQueueClosed = errors.New("queue: queue is closed")
ErrQueueClosed is returned when operating on a closed queue.
Functions ¶
func NewMemoryQueue ¶
NewMemoryQueue creates a thread-safe, priority-ordered in-memory queue.
Pop blocks until a job is available or the context is cancelled. Close unblocks all waiting Pop callers.
Types ¶
type RedisQueue ¶
type RedisQueue struct {
// contains filtered or unexported fields
}
RedisQueue implements foxhound.Queue using a Redis sorted set.
Jobs are scored so that higher priority always surfaces first. Within a priority tier, FIFO order is preserved by incorporating a microsecond timestamp into the score:
score = -(priority * 1_000_000_000) + created_at_micros
ZPOPMIN selects the member with the lowest score. Negating the priority component ensures higher-priority jobs have a more-negative base score and are therefore selected first. Adding (not subtracting) the microsecond timestamp as a positive offset means an older job (smaller micros) produces a lower overall score than a newer job within the same priority tier, giving correct FIFO ordering. Distributed workers can safely share the same key because ZPOPMIN is atomic.
func NewRedis ¶
func NewRedis(addr, password string, db int, queueKey string) (*RedisQueue, error)
NewRedis creates a RedisQueue that connects to the given Redis address. The connection is tested with a PING; an error is returned if Redis is unavailable.
func NewRedisFromClient ¶
func NewRedisFromClient(client *redis.Client, queueKey string) *RedisQueue
NewRedisFromClient creates a RedisQueue from an existing redis.Client. Ownership of the client is transferred to the queue; Close will close it.
func (*RedisQueue) Close ¶
func (q *RedisQueue) Close() error
Close releases the underlying Redis client.
func (*RedisQueue) Len ¶
func (q *RedisQueue) Len() int
Len returns the number of jobs in the sorted set.
type ReliableConfig ¶
type ReliableConfig struct {
// MaxRetries is the maximum number of times a job can fail before moving to DLQ.
// Default: 3
MaxRetries int
// AckTimeout is how long a job can be "in-flight" before being re-queued.
// Default: 5 minutes
AckTimeout time.Duration
// StaleCheckInterval is how often to check for stale in-flight jobs.
// Default: 30 seconds
StaleCheckInterval time.Duration
}
ReliableConfig configures the reliable queue wrapper.
func DefaultReliableConfig ¶
func DefaultReliableConfig() ReliableConfig
DefaultReliableConfig returns sensible defaults.
type ReliableQueue ¶
type ReliableQueue struct {
// contains filtered or unexported fields
}
ReliableQueue wraps any foxhound.Queue with acknowledgment, retry, and dead-letter queue semantics. Jobs that fail multiple times are moved to the DLQ instead of being retried forever.
func NewReliable ¶
func NewReliable(inner foxhound.Queue, cfg ReliableConfig) *ReliableQueue
NewReliable wraps an existing Queue with reliable delivery semantics.
func (*ReliableQueue) Ack ¶
func (rq *ReliableQueue) Ack(job *foxhound.Job)
Ack acknowledges successful processing of a job.
func (*ReliableQueue) Close ¶
func (rq *ReliableQueue) Close() error
Close stops the stale recovery goroutine and closes the inner queue.
func (*ReliableQueue) DLQ ¶
func (rq *ReliableQueue) DLQ() []*foxhound.Job
DLQ returns all jobs in the dead letter queue.
func (*ReliableQueue) DLQLen ¶
func (rq *ReliableQueue) DLQLen() int
DLQLen returns the number of jobs in the dead letter queue.
func (*ReliableQueue) InFlightLen ¶
func (rq *ReliableQueue) InFlightLen() int
InFlightLen returns the number of jobs currently being processed.
func (*ReliableQueue) Len ¶
func (rq *ReliableQueue) Len() int
Len returns the number of pending jobs in the inner queue.
func (*ReliableQueue) Nack ¶
func (rq *ReliableQueue) Nack(ctx context.Context, job *foxhound.Job)
Nack signals that job processing failed. The job is re-queued or moved to DLQ.
type SQLiteQueue ¶
type SQLiteQueue struct {
// contains filtered or unexported fields
}
SQLiteQueue implements foxhound.Queue using SQLite for durability.
Jobs survive process restarts, making this queue suitable for the resume capability described in the architecture docs. Jobs are stored in a single table and moved from status='pending' to status='processing' on Pop so that concurrent callers each see a distinct job.
func NewSQLite ¶
func NewSQLite(dbPath string) (*SQLiteQueue, error)
NewSQLite opens (or creates) a SQLite database at dbPath and initialises the jobs table.
func (*SQLiteQueue) Close ¶
func (q *SQLiteQueue) Close() error
Close closes the underlying database connection.
func (*SQLiteQueue) Len ¶
func (q *SQLiteQueue) Len() int
Len returns the number of pending jobs (status='pending').
Source Files