queue

Documentation

Index

• type Queue
  • func NewQueue(store storage.Store, cfg *config.Config, ...) *Queue
  • func (q *Queue) Ack(id uint64) error
  • func (q *Queue) Create(queueType, queueName string, settings entity.QueueSettings) error
  • func (q *Queue) Delete(id uint64) error
  • func (q *Queue) DeleteQueue() error
  • func (q *Queue) Dequeue(ack bool) (*entity.Message, error)
  • func (q *Queue) Enqueue(id uint64, group string, priority int64, content string, ...) (*entity.Message, error)
  • func (q *Queue) Get(id uint64) (*entity.Message, error)
  • func (q *Queue) GetStats() *QueueInfo
  • func (q *Queue) Init(queueType, queueName string, settings entity.QueueSettings) error
  • func (q *Queue) Len() int
  • func (q *Queue) Load(queueName string) error
  • func (q *Queue) Nack(id uint64, priority int64, metadata map[string]string) error
  • func (q *Queue) PersistSnapshot(sink raft.SnapshotSink) error
  • func (q *Queue) StartAckQueueMonitoring()
  • func (q *Queue) StopAckQueueMonitoring()
  • func (q *Queue) UpdatePriority(id uint64, newPriority int64) error
• type QueueInfo
• type QueueManager
  • func NewQueueManager(store storage.Store, cfg *config.Config, metrics *metrics.PrometheusMetrics) *QueueManager
  • func (qm *QueueManager) CreateQueue(queueType, queueName string, settings entity.QueueSettings) (*Queue, error)
  • func (qm *QueueManager) DeleteQueue(queueName string) error
  • func (qm *QueueManager) GetQueue(queueName string) (*Queue, error)
  • func (qm *QueueManager) GetQueues() []*QueueInfo
  • func (qm *QueueManager) PersistSnapshot(sink raft.SnapshotSink) error

Types

type Queue

type Queue struct {
	PrometheusMetrics *metrics.PrometheusMetrics
	// contains filtered or unexported fields
}

func NewQueue

func NewQueue(
	store storage.Store, cfg *config.Config, promMetrics *metrics.PrometheusMetrics,
) *Queue

func (*Queue) Ack

func (q *Queue) Ack(id uint64) error

func (*Queue) Create

func (q *Queue) Create(queueType, queueName string, settings entity.QueueSettings) error

func (*Queue) Delete

func (q *Queue) Delete(id uint64) error

func (*Queue) DeleteQueue

func (q *Queue) DeleteQueue() error

func (*Queue) Dequeue

func (q *Queue) Dequeue(ack bool) (*entity.Message, error)

func (*Queue) Enqueue

func (q *Queue) Enqueue(
	id uint64,
	group string,
	priority int64,
	content string,
	metadata map[string]string,
) (*entity.Message, error)

func (*Queue) Get

func (q *Queue) Get(id uint64) (*entity.Message, error)

func (*Queue) GetStats

func (q *Queue) GetStats() *QueueInfo

func (*Queue) Init

func (q *Queue) Init(queueType, queueName string, settings entity.QueueSettings) error

func (*Queue) Len

func (q *Queue) Len() int

func (*Queue) Load

func (q *Queue) Load(queueName string) error

func (*Queue) Nack

func (q *Queue) Nack(id uint64, priority int64, metadata map[string]string) error

func (*Queue) PersistSnapshot

func (q *Queue) PersistSnapshot(sink raft.SnapshotSink) error

func (*Queue) StartAckQueueMonitoring

func (q *Queue) StartAckQueueMonitoring()

func (*Queue) StopAckQueueMonitoring

func (q *Queue) StopAckQueueMonitoring()

func (*Queue) UpdatePriority

func (q *Queue) UpdatePriority(id uint64, newPriority int64) error

type QueueInfo

type QueueInfo struct {
	Name     string
	Type     string
	Settings entity.QueueSettings
	Stats    *metrics.Stats
	Ready    int64
	Unacked  int64
	Total    int64
}

type QueueManager

type QueueManager struct {
	PrometheusMetrics *metrics.PrometheusMetrics
	// contains filtered or unexported fields
}

func NewQueueManager

func NewQueueManager(store storage.Store, cfg *config.Config, metrics *metrics.PrometheusMetrics) *QueueManager

func (*QueueManager) CreateQueue

func (qm *QueueManager) CreateQueue(queueType, queueName string, settings entity.QueueSettings) (*Queue, error)

func (*QueueManager) DeleteQueue

func (qm *QueueManager) DeleteQueue(queueName string) error

func (*QueueManager) GetQueue

func (qm *QueueManager) GetQueue(queueName string) (*Queue, error)

func (*QueueManager) GetQueues

func (qm *QueueManager) GetQueues() []*QueueInfo

func (*QueueManager) PersistSnapshot

func (qm *QueueManager) PersistSnapshot(sink raft.SnapshotSink) error