bigqueue

Jan 20, 2020

Changes in this version

+ var ErrDifferentQueues = errors.New("consumers from different queues")

+ var ErrIncompatibleVersion = errors.New("incompatible format of the code and data")

+ type Consumer struct

+ func (c *Consumer) Dequeue() ([]byte, error)

+ func (c *Consumer) DequeueString() (string, error)

+ func (c *Consumer) IsEmpty() bool

type MmapQueue

+ func (q *MmapQueue) DequeueString() (string, error)

+ func (q *MmapQueue) FromConsumer(name string, from *Consumer) (*Consumer, error)

+ func (q *MmapQueue) NewConsumer(name string) (*Consumer, error)

Apr 10, 2019

Changes in this version

+ var ErrMustBeGreaterThanZero = errors.New("must be greater than zero")

+ type MmapQueue struct

+ func (q *MmapQueue) Close() error

+ func (q *MmapQueue) Dequeue() error

+ func (q *MmapQueue) Enqueue(message []byte) error

+ func (q *MmapQueue) EnqueueString(message string) error

+ func (q *MmapQueue) Flush() error

+ func (q *MmapQueue) IsEmpty() bool

+ func (q *MmapQueue) Peek() ([]byte, error)

+ func (q *MmapQueue) PeekString() (string, error)

type Option

+ func SetPeriodicFlushDuration(flushPeriod time.Duration) Option

+ func SetPeriodicFlushOps(flushMutOps int64) Option

+ type Queue interface

+ Close func() error

+ Dequeue func() error

+ Enqueue func([]byte) error

+ EnqueueString func(string) error

+ Flush func() error

+ IsEmpty func() bool

+ Peek func() ([]byte, error)

+ PeekString func() (string, error)

+ func NewMmapQueue(dir string, opts ...Option) (Queue, error)

Mar 10, 2019

Changes in this version

+ var ErrInvalidArenaSize = errors.New("mismatch in arena size")

+ var ErrTooFewInMemArenas = errors.New("too few in memory arenas")

+ var ErrTooSmallArenaSize = errors.New("too small arena size")

+ type Option func(*bqConfig) error

+ func SetArenaSize(arenaSize int) Option

+ func SetMaxInMemArenas(maxInMemArenas int) Option

Oct 14, 2018

Changes in this version

+ var ErrEmptyQueue = errors.New("queue is empty")

+ type Arena struct

+ func NewArena(file string, size int) (*Arena, error)

+ type BigQueue struct

+ func NewBigQueue(dir string) (*BigQueue, error)

+ func (bq *BigQueue) Close()

+ func (bq *BigQueue) Dequeue() ([]byte, error)

+ func (bq *BigQueue) Enqueue(message []byte) error

+ func (bq *BigQueue) IsEmpty() bool

+ func (bq *BigQueue) Peek() ([]byte, error)

+ type IBigQueue interface

+ Close func()

+ Dequeue func() ([]byte, error)

+ Enqueue func(elem []byte) error

+ IsEmpty func() bool

+ Peek func() []byte

+ type QueueIndex struct

+ func NewQueueIndex(dataDir string) (*QueueIndex, error)

+ func (i *QueueIndex) Flush()

+ func (i *QueueIndex) GetHead() (int, int)

+ func (i *QueueIndex) GetTail() (int, int)

+ func (i *QueueIndex) UpdateHead(aid, pos int)

+ func (i *QueueIndex) UpdateTail(aid, pos int)