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
Directories