workers

README

Sidekiq compatible background workers in golang.

• reliable queueing for all queues using brpoplpush
• handles retries
• support custom middleware
• customize concurrency per queue
• responds to Unix signals to safely wait for jobs to finish before exiting.
• provides stats on what jobs are currently running
• well tested

Example usage:

package main

import (
	"github.com/topfreegames/go-workers"
	workers "go-workers"
)

func myJob(message *workers.Msg) {
	// do something with your message
	// message.Jid()
	// message.Args() is a wrapper around go-simplejson (http://godoc.org/github.com/bitly/go-simplejson)
}

type myMiddleware struct{}

func (r *myMiddleware) Call(queue string, message *workers.Msg, next func() bool) (acknowledge bool) {
	// do something before each message is processed
	acknowledge = next()
	// do something after each message is processed
	return
}

func main() {
	workers.Configure(workers.Options{
		// location of redis instance
		Address: "localhost:6379",
		// instance of the database
		Database: "0",
		// number of connections to keep open with redis
		PoolSize: "30",
		// unique process id for this instance of workers (for proper recovery of inprogress jobs on crash)
		ProcessID: "1",
	})

	workers.Middleware.Append(&myMiddleware{})

	// pull messages from "myqueue" with concurrency of 10
	workers.Process("myqueue", myJob, 10)

	// pull messages from "myqueue2" with concurrency of 20
	workers.Process("myqueue2", myJob, 20)

	// Add a job to a queue
	workers.Enqueue("myqueue3", "Add", []int{1, 2})

	// Add a job to a queue with retry
	workers.EnqueueWithOptions("myqueue3", "Add", []int{1, 2}, workers.EnqueueOptions{Retry: true})

	// Add a job to a queue in a different redis instance
	workers.EnqueueWithOptions("myqueue4", "Add", []int{1, 2},
		workers.EnqueueOptions{
			Retry: true,
			ConnectionOptions: workers.Options{
				Address:  "localhost:6378",
				Database: "my-database",
				PoolSize: 10,
				Password: "pass",
			},
		},
	)

	// stats will be available at http://localhost:8080/stats
	go workers.StatsServer(8080)

	// Blocks until process is told to exit via unix signal
	workers.Run()
}

Initial development sponsored by Customer.io

Documentation

Index

• Constants
• Variables
• func BeforeStart(f func())
• func Configure(options Options)
• func DuringDrain(f func())
• func Enqueue(queue, class string, args interface{}) (string, error)
• func EnqueueAt(queue, class string, at time.Time, args interface{}) (string, error)
• func EnqueueIn(queue, class string, in float64, args interface{}) (string, error)
• func EnqueueWithOptions(queue, class string, args interface{}, opts EnqueueOptions) (string, error)
• func GetConnectionPool(options Options) *redis.Pool
• func Process(queue string, job jobFunc, concurrency int, mids ...Action)
• func Quit()
• func ResetManagers() error
• func Run()
• func SetLogger(l WorkersLogger)
• func Start()
• func Stats(w http.ResponseWriter, req *http.Request)
• func StatsServer(port int)
• type Action
• type Args
  • func (d Args) Equals(other interface{}) bool
  • func (d Args) ToJson() string
• type EnqueueData
• type EnqueueOptions
• type Fetcher
  • func NewFetch(queue string, messages chan *Msg, ready chan bool) Fetcher
• type MiddlewareLogging
  • func (l *MiddlewareLogging) Call(queue string, message *Msg, next func() bool) (acknowledge bool)
• type MiddlewareRetry
  • func (r *MiddlewareRetry) Call(queue string, message *Msg, next func() bool) (acknowledge bool)
• type MiddlewareStats
  • func (l *MiddlewareStats) Call(queue string, message *Msg, next func() bool) (acknowledge bool)
• type Middlewares
  • func NewMiddleware(actions ...Action) *Middlewares
  • func (m *Middlewares) Append(action Action)
  • func (m *Middlewares) Prepend(action Action)
• type Msg
  • func NewMsg(content string) (*Msg, error)
  • func (m *Msg) Args() *Args
  • func (d Msg) Equals(other interface{}) bool
  • func (m *Msg) Jid() string
  • func (m *Msg) OriginalJson() string
  • func (d Msg) ToJson() string
• type Options
• type RetryOptions
• type WorkerConfig
• type WorkerStats
  • func GetStats() *WorkerStats
• type WorkersLogger

Constants

const (
	DEFAULT_MAX_RETRY = 25
	LAYOUT            = "2006-01-02 15:04:05 MST"
)

const (
	RETRY_KEY          = "goretry"
	SCHEDULED_JOBS_KEY = "schedule"
)

const (
	NanoSecondPrecision = 1000000000.0
)

Variables

var Logger = initLogger()

Logger is the default logger

var Middleware = NewMiddleware(
	&MiddlewareLogging{},
	&MiddlewareRetry{},
	&MiddlewareStats{},
)

Functions

func BeforeStart

func BeforeStart(f func())

func Configure

func Configure(options Options)

func DuringDrain

func DuringDrain(f func())

func Enqueue

func Enqueue(queue, class string, args interface{}) (string, error)

func EnqueueAt

func EnqueueAt(queue, class string, at time.Time, args interface{}) (string, error)

func EnqueueIn

func EnqueueIn(queue, class string, in float64, args interface{}) (string, error)

func EnqueueWithOptions

func EnqueueWithOptions(queue, class string, args interface{}, opts EnqueueOptions) (string, error)

func GetConnectionPool

func GetConnectionPool(options Options) *redis.Pool

func Process

func Process(queue string, job jobFunc, concurrency int, mids ...Action)

func Quit

func Quit()

func ResetManagers

func ResetManagers() error

func Run

func Run()

func SetLogger

func SetLogger(l WorkersLogger)

SetLogger rewrites the default logger

func Start

func Start()

func Stats

func Stats(w http.ResponseWriter, req *http.Request)

Stats writes stats on response writer

func StatsServer

func StatsServer(port int)

Types

type Action

type Action interface {
	Call(queue string, message *Msg, next func() bool) bool
}

type Args

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

func (Args) Equals

func (d Args) Equals(other interface{}) bool

func (Args) ToJson

func (d Args) ToJson() string

type EnqueueData

type EnqueueData struct {
	Queue      string      `json:"queue,omitempty"`
	Class      string      `json:"class"`
	Args       interface{} `json:"args"`
	Jid        string      `json:"jid"`
	EnqueuedAt float64     `json:"enqueued_at"`
	EnqueueOptions
}

type EnqueueOptions

type EnqueueOptions struct {
	RetryCount        int          `json:"retry_count,omitempty"`
	Retry             bool         `json:"retry,omitempty"`
	RetryMax          int          `json:"retry_max,omitempty"`
	At                float64      `json:"at,omitempty"`
	RetryOptions      RetryOptions `json:"retry_options,omitempty"`
	ConnectionOptions Options      `json:"connection_options,omitempty"`
}

type Fetcher

type Fetcher interface {
	Queue() string
	Fetch()
	Acknowledge(*Msg)
	Ready() chan bool
	FinishedWork() chan bool
	Messages() chan *Msg
	Close()
	Closed() bool
}

func NewFetch

func NewFetch(queue string, messages chan *Msg, ready chan bool) Fetcher

type MiddlewareLogging

type MiddlewareLogging struct{}

func (*MiddlewareLogging) Call

func (l *MiddlewareLogging) Call(queue string, message *Msg, next func() bool) (acknowledge bool)

type MiddlewareRetry

type MiddlewareRetry struct{}

func (*MiddlewareRetry) Call

func (r *MiddlewareRetry) Call(queue string, message *Msg, next func() bool) (acknowledge bool)

type MiddlewareStats

type MiddlewareStats struct{}

func (*MiddlewareStats) Call

func (l *MiddlewareStats) Call(queue string, message *Msg, next func() bool) (acknowledge bool)

type Middlewares

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

func NewMiddleware

func NewMiddleware(actions ...Action) *Middlewares

func (*Middlewares) Append

func (m *Middlewares) Append(action Action)

func (*Middlewares) Prepend

func (m *Middlewares) Prepend(action Action)

type Msg

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

func NewMsg

func NewMsg(content string) (*Msg, error)

func (*Msg) Args

func (m *Msg) Args() *Args

func (Msg) Equals

func (d Msg) Equals(other interface{}) bool

func (*Msg) Jid

func (m *Msg) Jid() string

func (*Msg) OriginalJson

func (m *Msg) OriginalJson() string

func (Msg) ToJson

func (d Msg) ToJson() string

type Options

type Options struct {
	Address      string
	Password     string
	Database     string
	ProcessID    string
	Namespace    string
	PoolSize     int
	PoolInterval int
	DialOptions  []redis.DialOption
}

type RetryOptions

type RetryOptions struct {
	Exp      int `json:"exp"`
	MinDelay int `json:"min_delay"`
	MaxDelay int `json:"max_delay"`
	MaxRand  int `json:"max_rand"`
}

type WorkerConfig

type WorkerConfig struct {
	Namespace    string
	PoolInterval int
	Pool         *redis.Pool
	Fetch        func(queue string) Fetcher
	// contains filtered or unexported fields
}

var Config *WorkerConfig

type WorkerStats

type WorkerStats struct {
	Processed int               `json:"processed"`
	Failed    int               `json:"failed"`
	Enqueued  map[string]string `json:"enqueued"`
	Retries   int64             `json:"retries"`
}

WorkerStats holds workers stats

func GetStats

func GetStats() *WorkerStats

GetStats returns workers stats

type WorkersLogger

type WorkersLogger interface {
	Fatal(format ...interface{})
	Fatalf(format string, args ...interface{})
	Fatalln(args ...interface{})

	Debug(args ...interface{})
	Debugf(format string, args ...interface{})
	Debugln(args ...interface{})

	Error(args ...interface{})
	Errorf(format string, args ...interface{})
	Errorln(args ...interface{})

	Info(args ...interface{})
	Infof(format string, args ...interface{})
	Infoln(args ...interface{})

	Warn(args ...interface{})
	Warnf(format string, args ...interface{})
	Warnln(args ...interface{})
}

WorkersLogger represents the log interface