README
¶
Beanspike
Beanstalk inspired job queue backed by Aerospike KVS.
Basic usage
Put
conn, _ := DialDefault()
tube, _ := conn.Use("testtube")
id, err := tube.Put([]byte("hello"), 0, 0, true)
Put(body []byte, delay time.Duration, ttr time.Duration, lz bool) where delay will hold the job in a delayed mode before releasing it for processing. Note, in practice all delayed jobs will be behind any immediately available jobs in the tube. ttr is the time the consumer of the job must Touch the task within or have it return to the ready state. If lz is true, the action will try and compress the payload trading some performance for memory usage on Aerospike. Note, if the payload is too small or the LZ4 compression is unable to provide a space saving, the entry may be inserted uncompressed anyway. Passing true is largely safe from a performance prespective unless you know that body is large and high entorpy. Cases where compression was attempted and abandoned in currently present jobs are reflected in Stats.SkippedSize.
Reserve & Delete
conn, _ := DialDefault()
tube, _ := conn.Use("testtube")
id, body, ttl, err := tube.Reserve()
....
exists, err := tube.Delete(id)
Touch
err = tube.Touch(id)
If a job is Put in the tube with a ttr, the job must be touched at some period smaller than this value to keep it reserved. An error on Touch should be treated as an instruction to abandon the job because it has been deleted or has timed out.
Stats
conn, _ := DialDefault()
tube, _ := conn.Use("testtube")
stats, err := tube.Stats()
Delete a tube
This is mainly for unit test clean up but there may be a legitimate usecase.
conn, _ := DialDefault()
conn.Delete("testtube")
Connection information
If DialDefault() is used, various reasonable defaults are set.
The client uses environment variables AEROSPIKE_HOST and AEROSPIKE_PORT to connect to an Aerospike node.
The library will attempt to parse AEROSPIKE_PORT either as a port number or a Docker environment variable as might be set if the the application is in a container that has been linked to an Aerospike conatiner with --link aerospike:aerospike.
If nothing is specified, the client assumes aerospike and 3000
The alternative Dial(id string, host string, port int) requires everything to be set explicitly.
Unit Tests
$ gpm
$ source gvp
$ go test
Benchmarks
$ go test --bench=.
...
BenchmarkPut 2000 589034 ns/op
BenchmarkReserve 200 14759261 ns/op
BenchmarkRelease 3000 607290 ns/op
TODO
- Fix TTL related implementations.
- Messaging implementation for
Reserve. - Possibly review stats. Review UDF lifecycle management.
KickandBuryunit tests
Documentation
¶
Overview ¶
Package bsjob is a wrapper around beanspike jobs
Index ¶
- Constants
- Variables
- type Client
- type Conn
- type Job
- func (job *Job) Bury(reason string) error
- func (job *Job) Delete() (bool, error)
- func (job *Job) Monitor() chan struct{}
- func (job *Job) Release(delay time.Duration) error
- func (job *Job) ReleaseWithRetry(delay time.Duration, incr, retryFlag bool) error
- func (job *Job) RetryCount() int
- func (job *Job) String() string
- func (job *Job) Touch() error
- type Stats
- type TaskHandler
- type Tube
- func (tube *Tube) Bury(id int64, reason []byte) error
- func (tube *Tube) Delete(id int64) (bool, error)
- func (tube *Tube) KickJob(id int64) error
- func (tube *Tube) PeekBuried() (id int64, body []byte, ttr time.Duration, reason []byte, err error)
- func (tube *Tube) Put(body []byte, delay time.Duration, ttr time.Duration, lz bool, metadata string) (id int64, err error)
- func (tube *Tube) Release(id int64, delay time.Duration) error
- func (tube *Tube) ReleaseWithRetry(id int64, delay time.Duration, incr, retryFlag bool) error
- func (tube *Tube) Reserve() (id int64, body []byte, ttr time.Duration, retries int, retryFlag bool, ...)
- func (tube *Tube) ReserveBatch(batchSize int) (jobs []*Job, err error)
- func (tube *Tube) Stats() (s *Stats, err error)
- func (tube *Tube) Touch(id int64) error
- type TubeID
Constants ¶
const ( IdleInTubeTime = 2 * time.Second OutTubeTTR = 30 * time.Second )
const ( AerospikeHost = "aerospike" AerospikePort = 3000 AerospikeHostEnv = "AEROSPIKE_HOST" AerospikePortEnv = "AEROSPIKE_PORT" AerospikeNamespace = "beanspike" CompressionSizeThreshold = 1024 * 2 // run admin operations at most 1/10 seconds AerospikeAdminDelay = 2 AerospikeAdminScanSize = 2500 AerospikeMetadataSet = "metadata" // Bin names. Don't exceed 14 chars AerospikeNameStatus = "status" AerospikeNameBody = "body" AerospikeNameBy = "by" AerospikeNameReason = "reason" AerospikeNameDelay = "delay" AerospikeNameDelayValue = "seconds" // these are not used, review AerospikeNameTtr = "ttr" AerospikeNameTtrKey = "ttrkey" AerospikeNameTtrValue = "seconds" // these are not used, review AerospikeNameCompressedSize = "csize" AerospikeNameSize = "size" AerospikeNameRetries = "retries" AerospikeNameRetryFlag = "retryflag" AerospikeNameMetadata = "metadata" AerospikeNameBuriedMetadata = "buried_meta" // For the scan policy AerospikeQueryQueueSize = 4 // For bin AerospikeNameStatus AerospikeSymReady = "READY" AerospikeSymReserved = "RESERVED" AerospikeSymReservedTtr = "RESERVEDTTR" AerospikeSymBuried = "BURIED" AerospikeSymDelayed = "DELAYED" AerospikeSymDeleted = "DELETED" AerospikeKeySuffixTtr = "reservedttr" AerospikeKeySuffixDelayed = "delayed" )
Variables ¶
var ( ErrAlreadyConnected = errors.New("already connected") ErrNotConnected = errors.New("not connected") )
var ( ErrEmptyRecord = errors.New("ASSERT: Record empty") ErrNotBuriedOrDelayed = errors.New("Job is not buried or delayed") ErrJobNotFound = errors.New("Job not found") )
var ErrParsing = errors.New("error parsing body")
Functions ¶
This section is empty.
Types ¶
type Conn ¶
type Conn struct {
// contains filtered or unexported fields
}
func DialDefaultWithRetry ¶
type TaskHandler ¶
type TaskHandler interface {
// ReserveAndDecode reserve a task and decode it to appropriate value type
ReserveAndDecode(*Tube) (*Job, interface{}, error)
// Handle processes incoming message.
// Calling context.CancelFunc stops all underlying goroutines and release a beanspike job
Handle(context.CancelFunc, interface{})
// OnRelease will be called when handler context canceled
OnRelease(interface{})
}
Source Files