Documentation
¶
Overview ¶
Package redis provides Redis-backed implementations of lagodev's broadcasting.Broadcaster and queue.Queue interfaces. Built on github.com/redis/go-redis/v9.
One Go module ships both drivers because they share the same Redis client dependency tree — install once, pick the type(s) you need:
go get github.com/devituz/lagodev/drivers/redis@latest
rdb := redislib.NewClient(&redislib.Options{Addr: "localhost:6379"})
b := redis.NewBroadcaster(rdb) // broadcasting.Broadcaster
q := redis.NewQueue(rdb, "default") // queue.Queue
Index ¶
- type Broadcaster
- type BroadcasterOption
- type Queue
- func (q *Queue) Ack(ctx context.Context, jobID string) error
- func (q *Queue) Len() int
- func (q *Queue) Nack(ctx context.Context, jobID string, retryAfter time.Duration) error
- func (q *Queue) Pop(ctx context.Context, wait time.Duration) (queue.Job, error)
- func (q *Queue) Purge(ctx context.Context) error
- func (q *Queue) Push(ctx context.Context, j queue.Job) error
- type QueueOption
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broadcaster ¶
type Broadcaster struct {
// contains filtered or unexported fields
}
Broadcaster fans events out over Redis Pub/Sub. One Redis client drives both Publish and Subscribe; each subscription holds its own goroutine reading from a PubSub.
func NewBroadcaster ¶
func NewBroadcaster(rdb *redislib.Client, opts ...BroadcasterOption) *Broadcaster
NewBroadcaster returns a Broadcaster bound to rdb.
func (*Broadcaster) Close ¶
func (b *Broadcaster) Close() error
Close cancels every active subscription and releases the client resources held by this Broadcaster. The shared *redis.Client is NOT closed (the caller owns it and may share it across drivers).
func (*Broadcaster) Dropped ¶
func (b *Broadcaster) Dropped() uint64
Dropped returns how many events were dropped due to handler errors being silently logged.
func (*Broadcaster) Publish ¶
func (b *Broadcaster) Publish(ctx context.Context, e broadcasting.Event) error
Publish encodes the event into a Redis PUBLISH frame ("<name>\n<payload>") so subscribers can recover the Name + Payload fields. The base lagodev Event has no Type other than Name and Payload, so this two-line encoding stays lossless.
func (*Broadcaster) Subscribe ¶
func (b *Broadcaster) Subscribe(ctx context.Context, channel string, fn broadcasting.Handler) (broadcasting.Subscription, error)
Subscribe spawns a goroutine that pumps the Redis PubSub channel into fn. The returned Subscription unbinds when Cancel is called.
type BroadcasterOption ¶
type BroadcasterOption func(*Broadcaster)
BroadcasterOption customises Broadcaster behaviour.
func WithLogger ¶
func WithLogger(fn func(string, ...any)) BroadcasterOption
WithLogger installs a structured logger called on handler errors or dropped events.
func WithPrefix ¶
func WithPrefix(prefix string) BroadcasterOption
WithPrefix prepends prefix + ":" to every channel name. Useful for running multiple lagodev apps against a shared Redis.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue implements queue.Queue against Redis.
Two Redis keys are used per logical queue:
- "<prefix>:queue:<name>" — LIST of ready jobs (LPUSH/BRPOPLPUSH)
- "<prefix>:queue:<name>:reserved" — sorted set of reserved jobs scored by their visibility- timeout deadline; reaper requeues entries past their deadline.
- "<prefix>:queue:<name>:delayed" — sorted set of delayed jobs scored by their available_at epoch.
Promotion (delayed → ready and reserved-timeout → ready) is handled inline on each Pop so no background worker is required.
func NewQueue ¶
func NewQueue(rdb *redislib.Client, name string, opts ...QueueOption) *Queue
NewQueue returns a Queue named name backed by rdb.
func (*Queue) Ack ¶
Ack removes the job from the reserved set. Linear scan is fine — reserved sets are small at any given moment (only in-flight jobs).
func (*Queue) Nack ¶
Nack returns the job to the queue. If retryAfter > 0, it's stored in the delayed zset; otherwise pushed back on the ready list. Attempts is incremented.
func (*Queue) Pop ¶
Pop polls Redis until a job is available or wait elapses, returning queue.ErrEmpty in the latter case.
type QueueOption ¶
type QueueOption func(*Queue)
QueueOption customises Queue behaviour.
func QueueWithPopPollInterval ¶
func QueueWithPopPollInterval(d time.Duration) QueueOption
QueueWithPopPollInterval controls how often Pop polls for new work (between BRPOPLPUSH timeouts). Default 100ms — lower values cut latency at the cost of Redis CPU.
func QueueWithPrefix ¶
func QueueWithPrefix(p string) QueueOption
QueueWithPrefix overrides the key prefix (default "lagodev").
func QueueWithVisibilityTimeout ¶
func QueueWithVisibilityTimeout(d time.Duration) QueueOption
QueueWithVisibilityTimeout sets how long a reserved job stays invisible. Default 5 minutes.
Source Files