msgqueue

Sep 26, 2017

Changes in this version

type Message

+ func (m *Message) EncodeArgs() (string, error)

Aug 14, 2017

Changes in this version

type Message

+ func (m *Message) EncodeBody() (string, error)

Aug 8, 2017

Changes in this version

type BatcherOptions

+ Handler func([]*Message) error

Aug 8, 2017

Changes in this version

type BatcherOptions

+ RetryLimit int

type Processor

+ func (p *Processor) Put(msg *Message) error

Aug 3, 2017

Changes in this version

+ func SetLogger(logger *log.Logger)

+ type Batcher struct

+ func NewBatcher(q Queue, opt *BatcherOptions) *Batcher

+ func (b *Batcher) Add(msg *Message)

+ func (b *Batcher) Close() error

+ func (b *Batcher) SetSync(v bool)

+ func (b *Batcher) Wait() error

+ type BatcherOptions struct

+ Splitter func([]*Message) ([]*Message, []*Message)

+ Timeout time.Duration

+ Worker func([]*Message) error

+ type Delayer interface

+ Delay func() time.Duration

+ type Manager interface

+ NewQueue func(*Options) Queue

+ Queues func() []Queue

type Message

+ Err error

+ func (m *Message) GetBody() (string, error)

type Options

+ PauseErrorsThreshold int

+ ReservationSize int

+ WaitTimeout time.Duration

+ type Processor struct

+ func NewProcessor(q Queue, opt *Options) *Processor

+ func StartProcessor(q Queue, opt *Options) *Processor

+ func (p *Processor) Add(msg *Message) error

+ func (p *Processor) Options() *Options

+ func (p *Processor) Process(msg *Message) error

+ func (p *Processor) ProcessAll() error

+ func (p *Processor) ProcessOne() error

+ func (p *Processor) Purge() error

+ func (p *Processor) Queue() Queue

+ func (p *Processor) SetNoDelete(flag bool) *Processor

+ func (p *Processor) Start() error

+ func (p *Processor) Stats() *ProcessorStats

+ func (p *Processor) Stop() error

+ func (p *Processor) StopTimeout(timeout time.Duration) error

+ func (p *Processor) String() string

+ type ProcessorStats struct

+ AvgDuration time.Duration

+ Buffered uint32

+ Fails uint32

+ InFlight uint32

+ MaxDuration time.Duration

+ MinDuration time.Duration

+ Processed uint32

+ Retries uint32

+ type Queue interface

+ Add func(msg *Message) error

+ Call func(args ...interface{}) error

+ CallOnce func(dur time.Duration, args ...interface{}) error

+ Close func() error

+ CloseTimeout func(timeout time.Duration) error

+ Delete func(msg *Message) error

+ Name func() string

+ Processor func() *Processor

+ Purge func() error

+ Release func(*Message) error

+ ReserveN func(n int) ([]*Message, error)

May 29, 2017

Changes in this version

type Options

+ MaxBackoff time.Duration

May 1, 2017

Changes in this version

type Options

+ WorkerLimit int

Apr 12, 2017

Changes in this version

type Redis

+ Subscribe func(channels ...string) *redis.PubSub

Apr 10, 2017

Changes in this version

type Options

+ MaxWorkers int

type Redis

+ Eval func(script string, keys []string, args ...interface{}) *redis.Cmd

Apr 5, 2017

Changes in this version

type Options

+ GroupName string

Apr 4, 2017

Changes in this version

+ var ErrDuplicate = errors.New("queue: message with such name already exists")

+ type Handler interface

+ HandleMessage func(msg *Message) error

+ func NewHandler(fn interface{}) Handler

+ type HandlerFunc func(*Message) error

+ func (fn HandlerFunc) HandleMessage(msg *Message) error

+ type Message struct

+ Args []interface{}

+ Body string

+ Delay time.Duration

+ Id string

+ Name string

+ ReservationId string

+ ReservedCount int

+ func NewMessage(args ...interface{}) *Message

+ func (m *Message) MarshalArgs() (string, error)

+ func (m *Message) SetDelayName(delay time.Duration, args ...interface{})

+ func (m *Message) String() string

+ type Options struct

+ BufferSize int

+ FallbackHandler interface{}

+ Handler interface{}

+ MinBackoff time.Duration

+ Name string

+ RateLimit timerate.Limit

+ RateLimiter RateLimiter

+ Redis Redis

+ ReservationTimeout time.Duration

+ RetryLimit int

+ ScavengerNumber int

+ Storage Storage

+ WorkerNumber int

+ func (opt *Options) Init()

+ type RateLimiter interface

+ AllowRate func(name string, limit timerate.Limit) (delay time.Duration, allow bool)

+ type Redis interface

+ Del func(...string) *redis.IntCmd

+ Pipelined func(func(pipe *redis.Pipeline) error) ([]redis.Cmder, error)

+ Publish func(channel, message string) *redis.IntCmd

+ SAdd func(key string, members ...interface{}) *redis.IntCmd

+ SMembers func(key string) *redis.StringSliceCmd

+ SetNX func(string, interface{}, time.Duration) *redis.BoolCmd

+ type Storage interface

+ Exists func(key string) bool