README
¶
Overview
Package redis is a redis backed implementation of the queue
interface
Installation
go get -u -v github.com/gopi-frame/queue/driver/redis
Import
import "github.com/gopi-frame/queue/redis"
Usage
package main
import (
"context"
"github.com/gopi-frame/queue"
"github.com/gopi-frame/queue/driver/redis"
redislib "github.com/redis/go-redis/v9"
)
type CustomJob struct {
queue.Job `json:"-"`
}
func (c *CustomJob) Handle() error {
// do something
return nil
}
func (c *CustomJob) Failed(err error) {
// handle failed job
}
func main() {
db := redislib.NewClient(&redis.Options{
Addr: "localhost:6379",
})
if err := db.Ping(context.Background()).Err(); err != nil {
panic(err)
}
q := redis.NewQueue(&redis.Config{
DB: db,
Name: "queue",
Job: new(CustomJob),
})
q.Enqueue(new(CustomJob))
q.Enqueue(new(CustomJob))
q.Enqueue(new(CustomJob))
fmt.Println("count:", q.Count()) // Output: count: 3
for {
if job, ok := q.Dequeue(); ok {
if err := job.Handle(); err != nil {
if job.GetQueueable().GetAttempts < job.GetMaxAttempts() {
q.Release(job)
} else {
job.Failed(err)
}
} else {
q.Ack(job)
}
} else {
time.Sleep(time.Millisecond * 100)
}
}
}
Documentation
¶
Overview ¶
Package redis is a redis queue driver for queue package.
Index ¶
- Constants
- func Open(options map[string]any) (queuecontract.Queue, error)
- type Config
- type Driver
- type Job
- type LuaScript
- type Option
- type Queue
- func (q *Queue) Ack(_ queuecontract.Job)
- func (q *Queue) Count() int64
- func (q *Queue) Dequeue() (queuecontract.Job, bool)
- func (q *Queue) Empty() bool
- func (q *Queue) Enqueue(job queuecontract.Job) (queuecontract.Job, bool)
- func (q *Queue) Name() string
- func (q *Queue) Release(job queuecontract.Job)
- func (q *Queue) Remove(job queuecontract.Job)
Constants ¶
const QueueJobIDKeyFormat = "GOPI:QUEUE:INDEX:%s"
QueueJobIDKeyFormat is the redis key format for storing job id in a stored set
const QueueJobItemKeyFormat = "GOPI:QUEUE:ITEMS:%s"
QueueJobItemKeyFormat is the redis key format for storing job hash in a hash table
Variables ¶
This section is empty.
Functions ¶
func Open ¶
func Open(options map[string]any) (queuecontract.Queue, error)
Open is a convenience function that calls Driver.Open.
Types ¶
type Config ¶
type Config struct {
DB rediscontract.Client
Name string
Job queuecontract.Job
}
Config for redis queue
func NewConfig ¶
func NewConfig(db rediscontract.Client, name string, job queuecontract.Job) *Config
NewConfig creates a new redis queue config
type Job ¶
type Job struct {
ID uuid.UUID `json:"id"`
Queue string `json:"queue"`
Payload queue.Job `json:"payload"`
Attempts int `json:"attempts"`
AvailableAt time.Time `json:"available_at"`
CreatedAt time.Time `json:"created_at"`
}
Job is redis queueable job wrapper
func (*Job) GetAttempts ¶
GetAttempts returns how many times the job has been attempted
type LuaScript ¶
type LuaScript struct{}
LuaScript is a Lua script set for redis queue
func (LuaScript) Dequeue ¶
Dequeue returns the Lua script for popping a job
Parameters:
KEYS[1] = index sorted set key KEYS[2] = hash table key ARGV[1] = number of jobs to pop ARGV[2] = current time
type Option ¶
func WithDB ¶
func WithDB(db rediscontract.Client) Option
WithDB sets the redis client. If db is nil, then db is not changed.
func WithJob ¶
func WithJob(job queuecontract.Job) Option
WithJob sets the job type. If job is nil, then an exception is returned.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue is a redis queue implementation
func (*Queue) Dequeue ¶
func (q *Queue) Dequeue() (queuecontract.Job, bool)
Dequeue removes and returns a job from the queue
func (*Queue) Enqueue ¶
func (q *Queue) Enqueue(job queuecontract.Job) (queuecontract.Job, bool)
Enqueue adds a job to the queue
func (*Queue) Release ¶
func (q *Queue) Release(job queuecontract.Job)
Release releases a job back to the queue
func (*Queue) Remove ¶
func (q *Queue) Remove(job queuecontract.Job)
Remove removes a job from the queue
Source Files