redis

Documentation

Overview

Package redis contains a provider which supplies the manager with jobs stored in a redis list.

Index

• Constants
• Variables
• func RedisFactory() provider.Provider
• type Redis
  • func NewRedis(url string, poolSize int, JobList string) (*Redis, error)
  • func (r *Redis) CheckMemory(max int64) bool
  • func (r *Redis) Close() error
  • func (r *Redis) ConfigStruct() interface{}
  • func (r *Redis) ConfirmJob(j job.Job) error
  • func (r *Redis) Drain(stopChan chan struct{})
  • func (r *Redis) Init(i interface{}) error
  • func (r *Redis) Name() string
  • func (r *Redis) RequestWork(num int, jobChan chan job.Job) error
  • func (r *Redis) Target() float64
  • func (r *Redis) WaitTime(target float64) time.Duration
• type RedisConfig
• type RedisJob
  • func (r *RedisJob) Config() *job.JobConfig
  • func (r *RedisJob) JobConfirmer() job.JobConfirmer
• type TmpSet
  • func NewTmpSet(prefix string) *TmpSet
  • func (t *TmpSet) ConfirmJob(j *RedisJob, r *Redis) error
  • func (t *TmpSet) GetOrphan(r *Redis, max int) ([]*RedisJob, error)
  • func (t *TmpSet) PopAndLock(r *Redis) (*RedisJob, error)

Constants

const (
	DEFAULT_POOL_SIZE = 2
	DEFAULT_MAX_IDLE  = 100
	DEFAULT_HOST      = "localhost"
	DEFAULT_PORT      = "6379"
	DEFAULT_JOB_LIST  = "job_list"
	TEMP_JOB_LIST     = "tmp_job_list"
)

const (
	TMP_JOB_LOCK_PREFIX = "tmp_job:lock:"
)

Variables

var (
	MAX_WAIT_TIME    = 10 * time.Second
	DEFAULT_TIMEOUT  = 10 * time.Second
	REDIS_INFO_ERROR = errors.New("redis: failed to parse redis info")
)

var (
	JOB_NOT_FOUND = errors.New("redis: job not found")
)

Functions

func RedisFactory

func RedisFactory() provider.Provider

RedisFactory constructs a new redis provider from a ProviderConfig and returns it along with any errors

Types

type Redis

type Redis struct {
	*sync.Mutex

	JobList string
	// contains filtered or unexported fields
}

Redis holds a pool of redis connections that are used to talk to the database

func NewRedis

func NewRedis(url string, poolSize int, JobList string) (*Redis, error)

NewRedis create a new redis connection provider

func (*Redis) CheckMemory

func (r *Redis) CheckMemory(max int64) bool

CheckMemory check to see if the memory limit for the redis server has been hit

func (*Redis) Close

func (r *Redis) Close() error

Close close all of the connections to redis

func (*Redis) ConfigStruct

func (r *Redis) ConfigStruct() interface{}

func (*Redis) ConfirmJob

func (r *Redis) ConfirmJob(j job.Job) error

ConfirmJob removes the job from the tmp list on the redis server, signifying success

func (*Redis) Drain

func (r *Redis) Drain(stopChan chan struct{})

Drain pull jobs off redis and wrap them in a file job to be written to disk. this can be done if redis starts running out of memory. Drain will continue to remove jobs from redis until it either: a : gets a signal on it's stop channel b : runs out of jobs in the redis cue c : the connection to redis is closed

func (*Redis) Init

func (r *Redis) Init(i interface{}) error

Init initilize the redis provider

func (*Redis) Name

func (r *Redis) Name() string

Name

func (*Redis) RequestWork

func (r *Redis) RequestWork(num int, jobChan chan job.Job) error

RequestWork request parse jobs and send them to the manager to be processed

func (*Redis) Target

func (r *Redis) Target() float64

func (*Redis) WaitTime

func (r *Redis) WaitTime(target float64) time.Duration

WaitTime given a target of jobs persecond, how long should the manager wait before asking for more work

type RedisConfig

type RedisConfig struct {
	Host        string  `json:"host" required:"true" description:"The host of the redis server to connect to"`
	Port        string  `json:"port" required:"true" description:"Port of the redis server to connect to."`
	JobList     string  `json:"job_list" required:"true" description:"The list in redis to pull jobs from."`
	DumpOnLimit bool    `` /* 192-byte string literal not displayed */
	MemLimit    string  `` /* 157-byte string literal not displayed */
	Target      float64 `json:"target" required:"false" description:"The target jobs per second for this jobs on this job_list."`
}

RedisConfig contains config options for a redis provider

type RedisJob

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

RedisJob contains information about a job provided by redis

func (*RedisJob) Config

func (r *RedisJob) Config() *job.JobConfig

Config return the JobConfig for this job

func (*RedisJob) JobConfirmer

func (r *RedisJob) JobConfirmer() job.JobConfirmer

JobConfirmer return this job's provider

type TmpSet

type TmpSet struct {
	*sync.Mutex
	// contains filtered or unexported fields
}

TmpSet is a set in redis which holds temporary data

func NewTmpSet

func NewTmpSet(prefix string) *TmpSet

NewTmpSet create and return a new TmpSet

func (*TmpSet) ConfirmJob

func (t *TmpSet) ConfirmJob(j *RedisJob, r *Redis) error

ConfirmJob confirms a job

func (*TmpSet) GetOrphan

func (t *TmpSet) GetOrphan(r *Redis, max int) ([]*RedisJob, error)

GetAllOrphan gets all of the orphaned jobs in the redis list

func (*TmpSet) PopAndLock

func (t *TmpSet) PopAndLock(r *Redis) (*RedisJob, error)

Get a single job and lock it