beanstalk

README

Beanstalk

Go client for beanstalkd.

Install

$ go get github.com/beanstalkd/go-beanstalk

Use

Produce jobs:

c, err := beanstalk.Dial("tcp", "127.0.0.1:11300")
id, err := c.Put([]byte("hello"), 1, 0, 120*time.Second)

Consume jobs:

c, err := beanstalk.Dial("tcp", "127.0.0.1:11300")
id, body, err := c.Reserve(5 * time.Second)

Documentation

Overview

Package beanstalk provides a client for the beanstalk protocol. See http://kr.github.com/beanstalkd/ for the server.

This package is synchronized internally and safe to use from multiple goroutines without other coordination.

Example (Put)

id, err := conn.Put([]byte("myjob"), 1, 0, time.Minute)
if err != nil {
	panic(err)
}
fmt.Println("job", id)

Example (PutOtherTube)

tube := beanstalk.NewTube(conn, "mytube")
id, err := tube.Put([]byte("myjob"), 1, 0, time.Minute)
if err != nil {
	panic(err)
}
fmt.Println("job", id)

Example (Reserve)

id, body, err := conn.Reserve(5 * time.Second)
if err != nil {
	panic(err)
}
fmt.Println("job", id)
fmt.Println(string(body))

Example (ReserveOtherTubeSet)

tubeSet := beanstalk.NewTubeSet(conn, "mytube1", "mytube2")
id, body, err := tubeSet.Reserve(10 * time.Hour)
if err != nil {
	panic(err)
}
fmt.Println("job", id)
fmt.Println(string(body))

Index

• Constants
• Variables
• type Conn
  • func Dial(network, addr string) (*Conn, error)
  • func DialTimeout(network, addr string, timeout time.Duration) (*Conn, error)
  • func NewConn(conn io.ReadWriteCloser) *Conn
  • func (c *Conn) Bury(id uint64, pri uint32) error
  • func (c *Conn) Close() error
  • func (c *Conn) Delete(id uint64) error
  • func (c *Conn) KickJob(id uint64) error
  • func (c *Conn) ListTubes() ([]string, error)
  • func (c *Conn) Peek(id uint64) (body []byte, err error)
  • func (c *Conn) Release(id uint64, pri uint32, delay time.Duration) error
  • func (c *Conn) ReserveJob(id uint64) (body []byte, err error)
  • func (c *Conn) Stats() (map[string]string, error)
  • func (c *Conn) StatsJob(id uint64) (map[string]string, error)
  • func (c *Conn) Touch(id uint64) error
• type ConnError
  • func (e ConnError) Error() string
  • func (e ConnError) Unwrap() error
• type NameError
  • func (e NameError) Error() string
  • func (e NameError) Unwrap() error
• type Tube
  • func NewTube(c *Conn, name string) *Tube
  • func (t *Tube) Kick(bound int) (n int, err error)
  • func (t *Tube) Pause(d time.Duration) error
  • func (t *Tube) PeekBuried() (id uint64, body []byte, err error)
  • func (t *Tube) PeekDelayed() (id uint64, body []byte, err error)
  • func (t *Tube) PeekReady() (id uint64, body []byte, err error)
  • func (t *Tube) Put(body []byte, pri uint32, delay, ttr time.Duration) (id uint64, err error)
  • func (t *Tube) Stats() (map[string]string, error)
• type TubeSet
  • func NewTubeSet(c *Conn, name ...string) *TubeSet
  • func (t *TubeSet) Reserve(timeout time.Duration) (id uint64, body []byte, err error)

Examples

• Package (Put)
• Package (PutOtherTube)
• Package (Reserve)
• Package (ReserveOtherTubeSet)

Constants

const DefaultDialTimeout = 10 * time.Second

DefaultDialTimeout is the time to wait for a connection to the beanstalk server.

const DefaultKeepAlivePeriod = 10 * time.Second

DefaultKeepAlivePeriod is the default period between TCP keepalive messages.

const NameChars = `\-+/;.$_()0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz`

NameChars are the allowed name characters in the beanstalkd protocol.

Variables

var (
	ErrBadFormat  = errors.New("bad command format")
	ErrBuried     = errors.New("buried")
	ErrDeadline   = errors.New("deadline soon")
	ErrDraining   = errors.New("draining")
	ErrInternal   = errors.New("internal error")
	ErrJobTooBig  = errors.New("job too big")
	ErrNoCRLF     = errors.New("expected CR LF")
	ErrNotFound   = errors.New("not found")
	ErrNotIgnored = errors.New("not ignored")
	ErrOOM        = errors.New("server is out of memory")
	ErrTimeout    = errors.New("timeout")
	ErrUnknown    = errors.New("unknown command")
)

Error messages returned by the server.

var (
	ErrEmpty   = errors.New("name is empty")
	ErrBadChar = errors.New("name has bad char") // contains a character not in NameChars
	ErrTooLong = errors.New("name is too long")
)

Name format errors. The Err field of NameError contains one of these.

Types

type Conn

type Conn struct {
	Tube
	TubeSet
	// contains filtered or unexported fields
}

A Conn represents a connection to a beanstalkd server. It consists of a default Tube and TubeSet as well as the underlying network connection. The embedded types carry methods with them; see the documentation of those types for details.

func Dial

func Dial(network, addr string) (*Conn, error)

Dial connects addr on the given network using net.DialTimeout with a default timeout of 10s and then returns a new Conn for the connection.

func DialTimeout

func DialTimeout(network, addr string, timeout time.Duration) (*Conn, error)

DialTimeout connects addr on the given network using net.DialTimeout with a supplied timeout and then returns a new Conn for the connection.

func NewConn

func NewConn(conn io.ReadWriteCloser) *Conn

NewConn returns a new Conn using conn for I/O.

func (*Conn) Bury

func (c *Conn) Bury(id uint64, pri uint32) error

Bury places the given job in a holding area in the job's tube and sets its priority to pri. The job will not be scheduled again until it has been kicked; see also the documentation of Kick.

func (*Conn) Close

func (c *Conn) Close() error

Close closes the underlying network connection.

func (*Conn) Delete

func (c *Conn) Delete(id uint64) error

Delete deletes the given job.

func (*Conn) KickJob

func (c *Conn) KickJob(id uint64) error

KickJob places the given job to the ready queue of the same tube where it currently belongs when the given job id exists and is in a buried or delayed state.

func (*Conn) ListTubes

func (c *Conn) ListTubes() ([]string, error)

ListTubes returns the names of the tubes that currently exist on the server.

func (*Conn) Peek

func (c *Conn) Peek(id uint64) (body []byte, err error)

Peek gets a copy of the specified job from the server.

func (*Conn) Release

func (c *Conn) Release(id uint64, pri uint32, delay time.Duration) error

Release tells the server to perform the following actions: set the priority of the given job to pri, remove it from the list of jobs reserved by c, wait delay seconds, then place the job in the ready queue, which makes it available for reservation by any client.

func (*Conn) ReserveJob

func (c *Conn) ReserveJob(id uint64) (body []byte, err error)

ReserveJob reserves the specified job by id from the server.

func (*Conn) Stats

func (c *Conn) Stats() (map[string]string, error)

Stats retrieves global statistics from the server.

func (*Conn) StatsJob

func (c *Conn) StatsJob(id uint64) (map[string]string, error)

StatsJob retrieves statistics about the given job.

func (*Conn) Touch

func (c *Conn) Touch(id uint64) error

Touch resets the reservation timer for the given job. It is an error if the job isn't currently reserved by c. See the documentation of Reserve for more details.

type ConnError

type ConnError struct {
	Conn *Conn
	Op   string
	Err  error
}

ConnError records an error message from the server and the operation and connection that caused it.

func (ConnError) Error

func (e ConnError) Error() string

func (ConnError) Unwrap

func (e ConnError) Unwrap() error

type NameError

type NameError struct {
	Name string
	Err  error
}

NameError indicates that a name was malformed and the specific error describing how.

func (NameError) Error

func (e NameError) Error() string

func (NameError) Unwrap

func (e NameError) Unwrap() error

type Tube

type Tube struct {
	Conn *Conn
	Name string
}

Tube represents tube Name on the server connected to by Conn. It has methods for commands that operate on a single tube.

func NewTube

func NewTube(c *Conn, name string) *Tube

NewTube returns a new Tube representing the given name.

func (*Tube) Kick

func (t *Tube) Kick(bound int) (n int, err error)

Kick takes up to bound jobs from the holding area and moves them into the ready queue, then returns the number of jobs moved. Jobs will be taken in the order in which they were last buried.

func (*Tube) Pause

func (t *Tube) Pause(d time.Duration) error

Pause pauses new reservations in t for time d.

func (*Tube) PeekBuried

func (t *Tube) PeekBuried() (id uint64, body []byte, err error)

PeekBuried gets a copy of the job in the holding area that would be kicked next by Kick.

func (*Tube) PeekDelayed

func (t *Tube) PeekDelayed() (id uint64, body []byte, err error)

PeekDelayed gets a copy of the delayed job that is next to be put in t's ready queue.

func (*Tube) PeekReady

func (t *Tube) PeekReady() (id uint64, body []byte, err error)

PeekReady gets a copy of the job at the front of t's ready queue.

func (*Tube) Put

func (t *Tube) Put(body []byte, pri uint32, delay, ttr time.Duration) (id uint64, err error)

Put puts a job into tube t with priority pri and TTR ttr, and returns the id of the newly-created job. If delay is nonzero, the server will wait the given amount of time after returning to the client and before putting the job into the ready queue.

func (*Tube) Stats

func (t *Tube) Stats() (map[string]string, error)

Stats retrieves statistics about tube t.

type TubeSet

type TubeSet struct {
	Conn *Conn
	Name map[string]bool
}

TubeSet represents a set of tubes on the server connected to by Conn. Name names the tubes represented.

func NewTubeSet

func NewTubeSet(c *Conn, name ...string) *TubeSet

NewTubeSet returns a new TubeSet representing the given names.

func (*TubeSet) Reserve

func (t *TubeSet) Reserve(timeout time.Duration) (id uint64, body []byte, err error)

Reserve reserves and returns a job from one of the tubes in t. If no job is available before time timeout has passed, Reserve returns a ConnError recording ErrTimeout.

Typically, a client will reserve a job, perform some work, then delete the job with Conn.Delete.