redis

README

Redis

This is the basic initialization of this backend:

import (
	"log"

	"github.com/pulchre/wingman/backend/redis"
)

func main() {
	backend, err := redis.Init(redis.Options{})
	if err != nil {
		log.Fatal(err)
	}
}

This is required both for the application and for the manager.

The options available are:

• A go-redis options struct.
• BlockingTimeout time.Duration - Timeout for blocking commands to Redis.

Internals

Here is an overview of the stages that a job goes through in the Redis backend.

1. PushJob: RPUSH onto the queue.
2. PopJob: BLPOP pops the next job off the queue. If no job is on the queue, it blocks until the either a job is pushed, the timeout is reached, or the context is cancelled.
3. LockJob: If the job has no LockKey or concurrency is less than 1, we return wingman.LockNotRequired (0). A job may have n concurrency. The locks are a Redis key (wingman🔒LOCK_KEY:LOCK-ID). We randomly try a lock until we exhaust all the locks or get one. If we cannot get a lock, we move it to the held queue (wingman:held:QUEUE).
4. ProcessJob: SET set the processing key to the job. (wingman:processing:JOB-ID)
5. ReleaseJob: DEL the lock key and LMOVE a job from the held queue to the front of the queue.
6. ClearJob: DEL the processing key
7. FailJob: DEL & SET. first it deletes the processing key, then sets a failed key to the job (wingman:failed:JOB-ID).

Documentation

Index

• Variables
• type Backend
  • func Init(options Options) (*Backend, error)
  • func (b Backend) ClearJob(jobID string) error
  • func (b *Backend) Close() (err error)
  • func (b Backend) FailJob(job *wingman.InternalJob) error
  • func (b Backend) FailedJobs() uint64
  • func (b Backend) IncFailedJobs()
  • func (b Backend) IncSuccessfulJobs()
  • func (b Backend) LockJob(job *wingman.InternalJob) (wingman.LockID, error)
  • func (b Backend) Peek(queue string) (*wingman.InternalJob, error)
  • func (b Backend) PopJob(ctx context.Context, queue string) (*wingman.InternalJob, error)
  • func (b Backend) ProcessJob(job *wingman.InternalJob) error
  • func (b Backend) PushInternalJob(job *wingman.InternalJob) error
  • func (b Backend) PushJob(job wingman.Job) error
  • func (b Backend) ReleaseJob(job *wingman.InternalJob) error
  • func (b Backend) Size(queue string) uint64
  • func (b Backend) SuccessfulJobs() uint64
• type Config
• type Options

Variables

var (
	ErrCouldNotLock = errors.New("could not lock job")
)

Types

type Backend

type Backend struct {
	*redis.Client
	// contains filtered or unexported fields
}

func Init

func Init(options Options) (*Backend, error)

func (Backend) ClearJob

func (b Backend) ClearJob(jobID string) error

func (*Backend) Close

func (b *Backend) Close() (err error)

Release any resources used

func (Backend) FailJob

func (b Backend) FailJob(job *wingman.InternalJob) error

func (Backend) FailedJobs

func (b Backend) FailedJobs() uint64

func (Backend) IncFailedJobs

func (b Backend) IncFailedJobs()

func (Backend) IncSuccessfulJobs

func (b Backend) IncSuccessfulJobs()

func (Backend) LockJob

func (b Backend) LockJob(job *wingman.InternalJob) (wingman.LockID, error)

func (Backend) Peek

func (b Backend) Peek(queue string) (*wingman.InternalJob, error)

func (Backend) PopJob

func (b Backend) PopJob(ctx context.Context, queue string) (*wingman.InternalJob, error)

func (Backend) ProcessJob

func (b Backend) ProcessJob(job *wingman.InternalJob) error

func (Backend) PushInternalJob

func (b Backend) PushInternalJob(job *wingman.InternalJob) error

func (Backend) PushJob

func (b Backend) PushJob(job wingman.Job) error

func (Backend) ReleaseJob

func (b Backend) ReleaseJob(job *wingman.InternalJob) error

func (Backend) Size

func (b Backend) Size(queue string) uint64

func (Backend) SuccessfulJobs

func (b Backend) SuccessfulJobs() uint64

type Config

type Config struct {
	BlockingTimeout time.Duration
}

type Options

type Options struct {
	redis.Options
	Config
}