goqless

package module
v0.0.0-...-a77a673 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 3, 2013 License: MIT Imports: 16 Imported by: 0

README

Go (golang) version of https://github.com/seomoz/qless. A redis job queue.

Goqless now works but not every feature is implemented or tested. Any reported issues or pull requests are welcome.

Documentation

Overview

reference: https://github.com/seomoz/qless-py

This worker does not model the way qless does it. I more or less modeled it after my own needs.

Index

Constants

This section is empty.

Variables

View Source 
var (
	JOBSTATES = []string{"stalled", "running", "scheduled", "depends", "recurring"}
)

Functions

func Bool 

func Bool(reply interface{}, err error) (bool, error)

Bool is a helper that converts a command reply to a boolean. If err is not equal to nil, then Bool returns false, err. Otherwise Bool converts the reply to boolean as follows: 

Reply type      Result
integer         value != 0, nil
bulk            strconv.ParseBool(reply) or r != "False", nil
nil             false, ErrNil
other           false, error

func NewWorker 

func NewWorker(cli *Client, queues []string, interval int)

Types

type Client 

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

func Dial 

func Dial(host, port string) (*Client, error)

func NewClient 

func NewClient(host, port string) *Client

func (*Client) Close 

func (c *Client) Close()

func (*Client) Completed 

func (c *Client) Completed(start, count int) ([]string, error)

func (*Client) Do 

func (c *Client) Do(name string, keysAndArgs ...interface{}) (interface{}, error)

func (*Client) Events 

func (c *Client) Events() *Events

func (*Client) Get 

func (c *Client) Get(jid string) (interface{}, error)

func (*Client) GetConfig 

func (c *Client) GetConfig(option string) string

config(0, 'get', [option])

func (*Client) GetJob 

func (c *Client) GetJob(jid string) (*Job, error)

func (*Client) GetRecurringJob 

func (c *Client) GetRecurringJob(jid string) (*RecurringJob, error)

func (*Client) Queue 

func (c *Client) Queue(name string) *Queue

func (*Client) Queues 

func (c *Client) Queues(name string) ([]*Queue, error)

Queues(0, now, [queue])

func (*Client) SetConfig 

func (c *Client) SetConfig(option string, value interface{})

config(0, 'set', option, value)

func (*Client) Tagged 

func (c *Client) Tagged(tag string, start, count int) (*TaggedReply, error)

func (*Client) Track 

func (c *Client) Track(jid string) (bool, error)

Track(0, 'track', jid, now, tag, ...) Track the jid

func (*Client) Tracked 

func (c *Client) Tracked() (string, error)

Track(0) Returns all the tracked jobs

func (*Client) UnsetConfig 

func (c *Client) UnsetConfig(option string)

config(0, 'unset', [option])

func (*Client) Untrack 

func (c *Client) Untrack(jid string) (bool, error)

Track(0, 'untrack', jid, now) Untrack the jid

type Events 

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

func NewEvents 

func NewEvents(host, port string) *Events

func (*Events) Listen 

func (e *Events) Listen() (chan interface{}, error)

func (*Events) Unsubscribe 

func (e *Events) Unsubscribe()

type History 

type History struct {
	Popped int64
	Q      string
	Put    int64
	Worker string
}

type Job 

type Job struct {
	Expires      int64
	Dependents   StringSlice
	Tracked      bool
	Tags         StringSlice
	Jid          string
	Retries      int
	Data         interface{}
	Queue        string
	State        string
	Remaining    int
	Failure      interface{}
	History      []History
	Dependencies interface{}
	Klass        string
	Priority     int
	Worker       string
	// contains filtered or unexported fields
}

func NewJob 

func NewJob(cli *Client) *Job

func (*Job) Cancel 

func (j *Job) Cancel()

Cancel(0, id) Cancels this job

func (*Job) Client 

func (j *Job) Client() *Client

func (*Job) Complete 

func (j *Job) Complete() (string, error)

Complete(0, jid, worker, queue, now, data, ['next', n, [('delay', d) | ('depends', '["jid1","jid2",...]')]) Completes this job returns state, error

func (*Job) Depend 

func (j *Job) Depend(jids ...interface{}) (string, error)

Depends(0, jid, 'on', [jid, [jid, [...]]])

func (*Job) Fail 

func (j *Job) Fail(typ, message string) (bool, error)

Fail(0, id, worker, type, message, now, [data]) Fail this job return success, error

func (*Job) Heartbeat 

func (j *Job) Heartbeat() (bool, error)

Heartbeat(0, id, worker, now, [data]) Heartbeats this job return success, error

func (*Job) Move 

func (j *Job) Move() (string, error)

Move this from it's current queue into another

func (*Job) Retry 

func (j *Job) Retry(delay int) (string, error)

Retry(0, jid, queue, worker, now, [delay])

func (*Job) SetClient 

func (j *Job) SetClient(cli *Client)

func (*Job) Tag 

func (j *Job) Tag(tags ...interface{}) (string, error)

Tag(0, 'add', jid, now, tag, [tag, ...])

func (*Job) Track 

func (j *Job) Track() (bool, error)

Track(0, 'track', jid, now, tag, ...) Track this job

func (*Job) Undepend 

func (j *Job) Undepend(jids ...interface{}) (string, error)

Depends(0, jid, 'off', ('all' | [jid, [jid, [...]]]))

func (*Job) Untag 

func (j *Job) Untag(tags ...interface{}) (string, error)

Tag(0, 'remove', jid, now, tag, [tag, ...])

func (*Job) Untrack 

func (j *Job) Untrack() (bool, error)

Track(0, 'untrack', jid, now) Untrack this job

type JobCallback 

type JobCallback func(*Job) error

type JobFunc 

type JobFunc func(*Job) error

type Lua 

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

func NewLua 

func NewLua(c redis.Conn) *Lua

func (*Lua) Do 

func (l *Lua) Do(name string, keysAndArgs ...interface{}) (interface{}, error)

calls a script with the arguments

func (*Lua) LoadScripts 

func (l *Lua) LoadScripts(path string) error

loads all the given scripts from the path

type Queue 

type Queue struct {
	Running   int
	Name      string
	Waiting   int
	Recurring int
	Depends   int
	Stalled   int
	Scheduled int
	// contains filtered or unexported fields
}

func NewQueue 

func NewQueue(cli *Client) *Queue

func (*Queue) CancelAll 

func (q *Queue) CancelAll()

Cancel all jobs in this queue

func (*Queue) Jobs 

func (q *Queue) Jobs(state string, start, count int) ([]string, error)

Jobs(0, ('stalled' | 'running' | 'scheduled' | 'depends' | 'recurring'), now, queue, [offset, [count]])

func (*Queue) Pause 

func (q *Queue) Pause()

Pause(0, name)

func (*Queue) Pop 

func (q *Queue) Pop(count int) ([]*Job, error)

Pop(1, queue, worker, count, now) Pops a job off the queue.

func (*Queue) Put 

func (q *Queue) Put(jid, klass string, data interface{}, delay, priority int, tags []string, retries int, depends []string) (string, error)

Put(1, queue, jid, klass, data, now, delay, [priority, p], [tags, t], [retries, r], [depends, '[...]']) Puts a job into the queue returns jid, error

func (*Queue) Recur 

func (q *Queue) Recur(jid, klass string, data interface{}, interval, offset, priority int, tags []string, retries int) (string, error)

Recur(0, 'on', queue, jid, klass, data, now, 'interval', second, offset, [priority p], [tags t], [retries r]) Put a recurring job in this queue

func (*Queue) SetClient 

func (q *Queue) SetClient(cli *Client)

func (*Queue) Unpause 

func (q *Queue) Unpause()

Unpause(0, name)

type RecurringJob 

type RecurringJob struct {
	Tags     StringSlice
	Jid      string
	Retries  int
	Data     interface{}
	Queue    string
	Interval int
	Count    int
	Klass    string
	Priority int
	// contains filtered or unexported fields
}

func NewRecurringJob 

func NewRecurringJob(cli *Client) *RecurringJob

func (*RecurringJob) Cancel 

func (r *RecurringJob) Cancel()

func (*RecurringJob) Tag 

func (r *RecurringJob) Tag(tags ...interface{})

func (*RecurringJob) Untag 

func (r *RecurringJob) Untag(tags ...interface{})

func (*RecurringJob) Update 

func (r *RecurringJob) Update(opts map[string]interface{})

example: job.Update(map[string]interface{}{"priority": 5}) options: 

priority int
retries int
interval int
data interface{}
klass string

type StringSlice 

type StringSlice []string

represents a string slice with special json unmarshalling

func (*StringSlice) UnmarshalJSON 

func (s *StringSlice) UnmarshalJSON(data []byte) error

type TaggedReply 

type TaggedReply struct {
	Total int
	Jobs  StringSlice
}

type Worker 

type Worker struct {
	Interval int // in time.Duration
	// contains filtered or unexported fields
}

func (*Worker) AddFunc 

func (w *Worker) AddFunc(name string, f JobFunc) error

func (*Worker) AddService 

func (w *Worker) AddService(name string, rcvr interface{}) error

Adds all the methods in the passed interface as job functions. Job names are in the form of: name.methodname

func (*Worker) Start 

func (w *Worker) Start() error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.