bucket

README

Features

• In-Memory or Redis based storage
• Concurrency-safe
• Put, Take, Count, DynamicFill, and many more primitives
• Designed for distributed systems
• Well tested
• Well documented

Install

go get github.com/b3ntly/bucket

Documentation

Can be found here

Notes

• Test coverage badge is stuck in some cache and is out of date, click the badge to see the actual current coverage

Changelog for version 0.2

• Abstracted storage to its own interface, see storage.go and redis.go for examples
• Added in-memory storage option
• Replaced storageOptions parameter with IStorage, allowing a single storage option to be shared between multiple buckets. This should make it much more efficient to have a large number of buckets, i.e. per logged in user.

Changelog for version 0.3

• Renamed the repository from distributed-token-bucket to bucket
• Moved storage interface and providers to the /storage subpackage
• Added unit testing to the /storage subpackage
• Added watchable.go and changed signatures of all async functions to return a watchable
• Fixed examples
• Added more documentation and comments

Changelog for version 0.4

• Shortened "constructor" names
• Default options
• Better "constructor" signatures
• bucket.DynamicFill()
• bucket.TakeAll()

Changelog for version 0.5

• Added full documentation via Hugo

Benchmarks

go test -bench .

Benchmarks were done on a 2.2GHz quad-core Intel Core i7 with 16GB of RAM.

Version 0.4

Memory

Benchmark	Operations	ns/op
BenchmarkBucket_Create-8	10000	715 ns/op
BenchmarkBucket_Take-8	30000	132 ns/op
BenchmarkBucket_Put-8	50000	142 ns/op

Redis

Benchmark	Operations	ns/op
BenchmarkBucket_Create-8	10000	98582 ns/op
BenchmarkBucket_Take-8	30000	47716 ns/op
BenchmarkBucket_Put-8	50000	31350 ns/op

Documentation

Index

• Variables
• type Bucket
  • func New(options *Options) (*Bucket, error)
  • func NewWithRedis(options *Options) (*Bucket, error)
  • func (bucket *Bucket) Count() (int, error)
  • func (bucket *Bucket) DynamicFill(rate int, interval chan time.Time) *Watchable
  • func (bucket *Bucket) Fill(rate int, interval time.Duration) *Watchable
  • func (bucket *Bucket) Put(amount int) error
  • func (bucket *Bucket) Take(tokensDesired int) error
  • func (bucket *Bucket) TakeAll() (int, error)
  • func (bucket *Bucket) Watch(tokens int, duration time.Duration) *Watchable
• type Options
• type Watchable
  • func NewWatchable() *Watchable
  • func (w *Watchable) Close(err error)
  • func (w *Watchable) Done() chan error

Variables

var (
	DefaultMemoryStore = &storage.MemoryStorage{}

	// Options which use redis as the storage back-end, defaults to the default redis options
	DefaultRedisStore = &storage.RedisStorage{Client: redis.NewClient(&redis.Options{Addr: ":6379"})}
)

Types

type Bucket

type Bucket struct {

	// the name of the bucket, may have implications for certain storage providers
	Name string
	// contains filtered or unexported fields
}

func New

func New(options *Options) (*Bucket, error)

Instantiate the bucket with a given storage object. Test the validity of the storage with storage.Ping().

If a name already exists its storage will be shared with the bucket that created it.

The storage provider might reject a bucket name (and return an error) if the value already assigned to a bucket is unexpected, for example if a name has a string value in redis (which may indicate it is reserved for something else in the database). Redis will also reject sharing a bucket name whose value is 0, this was a personal choice because I found myself incorrectly using bucket names I thought did not exist but actually did (from leftover tests).

func NewWithRedis

func NewWithRedis(options *Options) (*Bucket, error)

Create a bucket with Redis storage

func (*Bucket) Count

func (bucket *Bucket) Count() (int, error)

Return an integer count of a bucket's token value

func (*Bucket) DynamicFill

func (bucket *Bucket) DynamicFill(rate int, interval chan time.Time) *Watchable

Dynamic fill fills the bucket every time it reads from interval

func (*Bucket) Fill

func (bucket *Bucket) Fill(rate int, interval time.Duration) *Watchable

Start a ticker that will periodically set the token value to a given rate on the defined interval. Returns a Watchable object identical to bucket.Watch, and thus may be canceled and observed.

func (*Bucket) Put

func (bucket *Bucket) Put(amount int) error

Increment the token value by a given amount.

func (*Bucket) Take

func (bucket *Bucket) Take(tokensDesired int) error

Decrement the token value of a bucket if the number of tokens in the bucket is >= tokensDesired. It will return an error if not enough tokens exist.

func (*Bucket) TakeAll

func (bucket *Bucket) TakeAll() (int, error)

returns a conditional amount of tokens representing all the tokens

func (*Bucket) Watch

func (bucket *Bucket) Watch(tokens int, duration time.Duration) *Watchable

Attempt on a 500ms interval to call bucket.Take with a nil response. It returns an instance of Watchable from which the polling can be cancelled and errors or nil may be received. See ./examples/watchable.go to get an idea of how it works.

type Options

type Options struct {
	Storage  storage.Storage
	Name     string
	Capacity int
}

type Watchable

type Watchable struct {
	Success chan error

	Cancel chan error

	Failed chan error

	// The final observable which the user is likely to read from. Though it can only be fired once it is buffered
	// so that is may be ignored.
	Finished chan error
}

a basic structure from which to cancel or observe an asynchronous action

func NewWatchable

func NewWatchable() *Watchable

func (*Watchable) Close

func (w *Watchable) Close(err error)

Close all channels to prevent memory leaks.

func (*Watchable) Done

func (w *Watchable) Done() chan error

Return the intended final observable