beanstalk

README

Beanstalk

Go client for beanstalkd.

New home

This package has a new home at https://github.com/beanstalkd/beanstalk.

The code is still available here to use, for any clients that depend on this import path, but there is no development on this repo. Maintenance and development now happen in the beanstalkd organization.

Please use github.com/beanstalkd/beanstalk in new code.

Install

$ go get github.com/kr/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)

package main

import (
	"fmt"
	"github.com/kr/beanstalk"
	"time"
)

var conn, _ = beanstalk.Dial("tcp", "127.0.0.1:11300")

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

Example (PutOtherTube)

package main

import (
	"fmt"
	"github.com/kr/beanstalk"
	"time"
)

var conn, _ = beanstalk.Dial("tcp", "127.0.0.1:11300")

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

Example (Reserve)

package main

import (
	"fmt"
	"github.com/kr/beanstalk"
	"time"
)

var conn, _ = beanstalk.Dial("tcp", "127.0.0.1:11300")

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

Example (ReserveOtherTubeSet)

package main

import (
	"fmt"
	"github.com/kr/beanstalk"
	"time"
)

var conn, _ = beanstalk.Dial("tcp", "127.0.0.1:11300")

func main() {
	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 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) 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) 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
• type NameError
  • func (e NameError) Error() string
• type 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 NameChars = `\-+/;.$_()0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz`

Characters allowed in a name 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 to the given address on the given network using net.Dial 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) 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) 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

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

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 (*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.