README
¶
Go Queue
Multi backend queues for Golang
Install
go get -u github.com/nickalie/go-queue
Example of usage
package main
import (
"fmt"
"github.com/nickalie/go-queue"
"time"
)
func main() {
for i := 0; i < 5; i++ {
go consumer(i + 1)
}
producer()
}
func producer() {
i := 0
for {
i++
queue.Put("messages", fmt.Sprintf("message %d", i))
time.Sleep(time.Second)
}
}
func consumer(index int) {
for {
var message string
queue.Get("messages", &message)
fmt.Printf("Consumer %d got a message: %s\n", index, message)
time.Sleep(2 * time.Second)
}
}
Documentation
¶
Index ¶
- func Get(queueName string, v interface{}) error
- func Put(queueName string, value interface{}) error
- func RemoveQueue(queueName string) error
- func Use(value Backend)
- type AMQPBackend
- type Backend
- type BeanstalkBackend
- type BuntBackend
- func (b *BuntBackend) Close() error
- func (b *BuntBackend) Codec(c Codec) *BuntBackend
- func (b *BuntBackend) Get(queueName string, v interface{}) error
- func (b *BuntBackend) Interval(interval time.Duration) *BuntBackend
- func (b *BuntBackend) Put(queueName string, value interface{}) error
- func (b *BuntBackend) RemoveQueue(queueName string) error
- func (b *BuntBackend) TTL(ttl time.Duration) *BuntBackend
- type ChannelBackend
- func (b *ChannelBackend) Buffer(buffer int) *ChannelBackend
- func (b *ChannelBackend) Channels(channels map[string]chan interface{}) *ChannelBackend
- func (b *ChannelBackend) Get(queueName string, v interface{}) error
- func (b *ChannelBackend) Put(queueName string, value interface{}) error
- func (b *ChannelBackend) RemoveQueue(queueName string) error
- type Codec
- type FSBackend
- type GOBCodec
- type JSONCodec
- type RedisBackend
- type StompBackend
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RemoveQueue ¶
Types ¶
type AMQPBackend ¶
type AMQPBackend struct {
// contains filtered or unexported fields
}
AMQPBackend provides AMQP-based backend to manage queues. https://en.wikipedia.org/wiki/Advanced_Message_Queuing_Protocol Suitable for multi-host, multi-process and multithreaded environment
func NewAMQPBackend ¶
func NewAMQPBackend(url string) (*AMQPBackend, error)
NewAMQPBackend creates new AMQPBackend
func (*AMQPBackend) Codec ¶
func (b *AMQPBackend) Codec(c Codec) *AMQPBackend
Codec sets codec to encode/decode objects in queues. GOBCodec is default.
func (*AMQPBackend) Get ¶
func (b *AMQPBackend) Get(queueName string, v interface{}) error
Get removes the first element from a queue and put it in the value pointed to by v
func (*AMQPBackend) Put ¶
func (b *AMQPBackend) Put(queueName string, value interface{}) error
Put adds value to the end of a queue.
func (*AMQPBackend) RemoveQueue ¶
func (b *AMQPBackend) RemoveQueue(queueName string) error
type Backend ¶
type Backend interface {
Put(queueName string, value interface{}) error
Get(queueName string, value interface{}) error
RemoveQueue(queueName string) error
}
Backend defines interface to manage queues. ChannelBackend is default.
type BeanstalkBackend ¶
type BeanstalkBackend struct {
// contains filtered or unexported fields
}
BeanstalkBackend provides beanstalk-based backend to manage queues. Suitable for multi-host, multi-process and multithreaded environment
func NewBeanstalkBackend ¶
func NewBeanstalkBackend(addr string) (*BeanstalkBackend, error)
NewBeanstalkBackend creates new BeanstalkBackend
func (*BeanstalkBackend) Codec ¶
func (b *BeanstalkBackend) Codec(c Codec) *BeanstalkBackend
Codec sets codec to encode/decode objects in queues. GOBCodec is default.
func (*BeanstalkBackend) Get ¶
func (b *BeanstalkBackend) Get(queueName string, value interface{}) error
Get removes the first element from a queue and put it in the value pointed to by value
func (*BeanstalkBackend) Put ¶
func (b *BeanstalkBackend) Put(queueName string, value interface{}) error
Put adds value to the end of a queue.
func (*BeanstalkBackend) RemoveQueue ¶
func (b *BeanstalkBackend) RemoveQueue(queueName string) error
type BuntBackend ¶
type BuntBackend struct {
// contains filtered or unexported fields
}
BuntBackend provides BuntDB-based backend to manage queues. https://github.com/tidwall/buntdb Suitable for multithreaded single process environment
func NewBuntBackend ¶
func NewBuntBackend(path string) (*BuntBackend, error)
NewBuntBackend creates new BuntBackend.
func NewBuntBackendFromDB ¶
func NewBuntBackendFromDB(db *buntdb.DB) *BuntBackend
NewBuntBackendFromDB creates new BuntBackend from bunt.DB object.
func (*BuntBackend) Codec ¶
func (b *BuntBackend) Codec(c Codec) *BuntBackend
Codec sets codec to encode/decode objects in queues. GOBCodec is default.
func (*BuntBackend) Get ¶
func (b *BuntBackend) Get(queueName string, v interface{}) error
Get removes the first element from a queue and put it in the value pointed to by v
func (*BuntBackend) Interval ¶
func (b *BuntBackend) Interval(interval time.Duration) *BuntBackend
Interval sets interval to poll new queue element. Default value is one second.
func (*BuntBackend) Put ¶
func (b *BuntBackend) Put(queueName string, value interface{}) error
Put adds value to the end of a queue.
func (*BuntBackend) RemoveQueue ¶
func (b *BuntBackend) RemoveQueue(queueName string) error
func (*BuntBackend) TTL ¶
func (b *BuntBackend) TTL(ttl time.Duration) *BuntBackend
TTL sets message expiration time. Default is 0 - message will never expire
type ChannelBackend ¶
ChannelBackend uses go channels to manage queues Suitable for multithreaded single process environment
func NewChannelBackend ¶
func NewChannelBackend() *ChannelBackend
NewChannelBackend creates new channels backend
func (*ChannelBackend) Buffer ¶
func (b *ChannelBackend) Buffer(buffer int) *ChannelBackend
Buffer sets default buffer for new channels created by the backend. Default value is 1000.
func (*ChannelBackend) Channels ¶
func (b *ChannelBackend) Channels(channels map[string]chan interface{}) *ChannelBackend
Channels sets initial channels (queues). Key - queue name, value - go channel
func (*ChannelBackend) Get ¶
func (b *ChannelBackend) Get(queueName string, v interface{}) error
Get removes the first element from a queue and put it in the value pointed to by v
func (*ChannelBackend) Put ¶
func (b *ChannelBackend) Put(queueName string, value interface{}) error
Put adds value to the end of a queue.
func (*ChannelBackend) RemoveQueue ¶
func (b *ChannelBackend) RemoveQueue(queueName string) error
type Codec ¶
type Codec interface {
Marshal(v interface{}) ([]byte, error)
Unmarshal(data []byte, v interface{}) error
}
Codec define interface for codecs to encode/decode objects in queues
type FSBackend ¶
type FSBackend struct {
// contains filtered or unexported fields
}
FSBackend uses file system to manage queues. Suitable for multithreaded and multi-process environments.
func NewFSBackend ¶
NewFSBackend creates new FSBackend.
func (*FSBackend) Get ¶
Get removes the first element from a queue and put it in the value pointed to by v
func (*FSBackend) Interval ¶
Interval sets interval to poll new queue element. Default value is one second.
func (*FSBackend) RemoveQueue ¶
type GOBCodec ¶
type GOBCodec byte
GOBCodec provides GOB based Codec
type JSONCodec ¶
type JSONCodec byte
JSONCodec provides JSON based Codec
type RedisBackend ¶
type RedisBackend struct {
// contains filtered or unexported fields
}
RedisBackend provides redis-based backend to manage queues. Suitable for multi-host, multi-process and multithreaded environment
func NewRedisBackend ¶
func NewRedisBackend(redisURL string) (*RedisBackend, error)
NewRedisBackend creates new RedisBackend
func NewRedisBackendWithPool ¶
func NewRedisBackendWithPool(pool *redis.Pool) *RedisBackend
NewRedisBackendWithPool creates new RedisBackend with provided redis.Pool
func (*RedisBackend) Codec ¶
func (b *RedisBackend) Codec(c Codec) *RedisBackend
Codec sets codec to encode/decode objects in queues. GOBCodec is default.
func (*RedisBackend) Get ¶
func (b *RedisBackend) Get(queueName string, v interface{}) error
Get removes the first element from a queue and put it in the value pointed to by v
func (*RedisBackend) Put ¶
func (b *RedisBackend) Put(queueName string, value interface{}) error
Put adds value to the end of a queue.
func (*RedisBackend) RemoveQueue ¶
func (b *RedisBackend) RemoveQueue(queueName string) error
type StompBackend ¶
type StompBackend struct {
// contains filtered or unexported fields
}
StompBackend provides stomp-based backend to manage queues. Suitable for multi-host, multi-process and multithreaded environment
func NewStompBackend ¶
NewStompBackend creates new RedisBackend
func (*StompBackend) Codec ¶
func (b *StompBackend) Codec(c Codec) *StompBackend
Codec sets codec to encode/decode objects in queues. GOBCodec is default.
func (*StompBackend) Get ¶
func (b *StompBackend) Get(queueName string, v interface{}) error
Get removes the first element from a queue and put it in the value pointed to by v
func (*StompBackend) Put ¶
func (b *StompBackend) Put(queueName string, value interface{}) error
Put adds value to the end of a queue.
func (*StompBackend) RemoveQueue ¶
func (b *StompBackend) RemoveQueue(queueName string) error
Source Files