cgroup

Documentation

Index

• Variables
• type CGManager
  • func NewCGManager(storagePath string, source *queue.Queue) (*CGManager, error)
  • func (m *CGManager) Close()
  • func (m *CGManager) ConsumerGroup(name string) (*ConsumerGroup, error)
  • func (m *CGManager) ConsumerGroupIterator() <-chan cmap.Tuple
  • func (m *CGManager) DeleteConsumerGroup(name string) error
• type CGQueue
  • func CGQueueOpen(name string, dataDir string) (*CGQueue, error)
  • func (q *CGQueue) Close()
  • func (q *CGQueue) Drop()
  • func (q *CGQueue) Flush() error
  • func (q *CGQueue) Path() string
• type ConsumerGroup
  • func NewConsumerGroup(name string, source *queue.Queue, storage *leveldb.DB) (*ConsumerGroup, error)
  • func (cg *ConsumerGroup) Delete() error
  • func (cg *ConsumerGroup) Flush() error
  • func (cg *ConsumerGroup) GetNext() ([]byte, error)
  • func (cg *ConsumerGroup) IsEmpty() bool
  • func (cg *ConsumerGroup) Length() uint64
  • func (cg *ConsumerGroup) Peek() ([]byte, error)
  • func (cg *ConsumerGroup) PutBack(value []byte) error
  • func (cg *ConsumerGroup) Source() queue.Consumer
  • func (cg *ConsumerGroup) Stats() *queue.Stats

Variables

var ErrInvalidName = errors.New("cgroup: name is not alphanumeric")

ErrInvalidName is returned when consumer group name is not alphanumeric

Types

type CGManager

type CGManager struct {
	sync.Mutex
	// contains filtered or unexported fields
}

CGManager represents multiple consumer group manager

func NewCGManager

func NewCGManager(storagePath string,
	source *queue.Queue) (*CGManager, error)

NewCGManager initializes new consumer group manager

func (*CGManager) Close

func (m *CGManager) Close()

Close consumer group manager

func (*CGManager) ConsumerGroup

func (m *CGManager) ConsumerGroup(name string) (*ConsumerGroup, error)

ConsumerGroup returns queue interface for provided consumer group name

func (*CGManager) ConsumerGroupIterator

func (m *CGManager) ConsumerGroupIterator() <-chan cmap.Tuple

ConsumerGroupIterator iterates through existing consumer groups

func (*CGManager) DeleteConsumerGroup

func (m *CGManager) DeleteConsumerGroup(name string) error

DeleteConsumerGroup deletes specified consumer group

type CGQueue

type CGQueue struct {
	sync.Mutex
	Name string

	*queue.Queue
	*CGManager
	// contains filtered or unexported fields
}

CGQueue represents queue with multiple consumer groups

func CGQueueOpen

func CGQueueOpen(name string, dataDir string) (*CGQueue, error)

CGQueueOpen opens a queue with multiple consumer groups

func (*CGQueue) Close

func (q *CGQueue) Close()

Close closes the queue

func (*CGQueue) Drop

func (q *CGQueue) Drop()

Drop closes the queue and removes it's data directory

func (*CGQueue) Flush

func (q *CGQueue) Flush() error

Flush drops all queue data

func (*CGQueue) Path

func (q *CGQueue) Path() string

Path returns queue data directory path

type ConsumerGroup

type ConsumerGroup struct {
	sync.RWMutex
	Name string
	// contains filtered or unexported fields
}

ConsumerGroup represents a consumer group that reads from a source queue, stores its own cursor position and saves failed reliable reads in order to serve them to other consumers later

func NewConsumerGroup

func NewConsumerGroup(name string, source *queue.Queue,
	storage *leveldb.DB) (*ConsumerGroup, error)

NewConsumerGroup initializes a consumer group

func (*ConsumerGroup) Delete

func (cg *ConsumerGroup) Delete() error

Delete deletes all the data associated with consumer group

func (*ConsumerGroup) Flush

func (cg *ConsumerGroup) Flush() error

Flush resets consumer group

func (*ConsumerGroup) GetNext

func (cg *ConsumerGroup) GetNext() ([]byte, error)

GetNext returns next value for that particular consumer group

func (*ConsumerGroup) IsEmpty

func (cg *ConsumerGroup) IsEmpty() bool

IsEmpty returns false if thereis no more items for this consumer group

func (*ConsumerGroup) Length

func (cg *ConsumerGroup) Length() uint64

Length returns remaining number of items for consumer group

func (*ConsumerGroup) Peek

func (cg *ConsumerGroup) Peek() ([]byte, error)

Peek returns next value without updating the cursor

func (*ConsumerGroup) PutBack

func (cg *ConsumerGroup) PutBack(value []byte) error

PutBack returns failed item back so it can be served to next consumer

func (*ConsumerGroup) Source

func (cg *ConsumerGroup) Source() queue.Consumer

Source returns source queue Consumer interface

func (*ConsumerGroup) Stats

func (cg *ConsumerGroup) Stats() *queue.Stats

Stats returns stats struct