queue

Nov 6, 2025

Changes in this version

type Queue

+ func (q *Queue) Touch(id uint64) error

Oct 19, 2025

Changes in this version

type Queue

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

+ func (q *Queue) UpdateQueue(settings entity.QueueSettings) error

type QueueManager

+ func (qm *QueueManager) UpdateQueue(queueName string, settings entity.QueueSettings) error

Jun 13, 2025

Changes in this version

type QueueInfo

+ Settings entity.QueueSettings

Mar 5, 2025

Changes in this version

type Queue

+ PrometheusMetrics *metrics.PrometheusMetrics

+ func NewQueue(db *badger.DB, cfg *config.Config, promMetrics *metrics.PrometheusMetrics) *Queue

+ func (bpq *Queue) Ack(id uint64) error

+ func (bpq *Queue) Create(queueType, queueName string) error

+ func (bpq *Queue) DeleteQueue() error

+ func (bpq *Queue) GetMessagesKey(id uint64) []byte

+ func (bpq *Queue) GetQueueKey(queueName string) []byte

+ func (bpq *Queue) GetStats() *QueueInfo

+ func (bpq *Queue) Init(queueType, queueName string) error

+ func (bpq *Queue) Load(queueName string, loadMessages bool) error

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

+ func (bpq *Queue) PersistSnapshot(sink raft.SnapshotSink) error

+ func (bpq *Queue) StartAckQueueMonitoring()

+ func (bpq *Queue) StopAckQueueMonitoring()

Mar 5, 2025

Changes in this version

type PriorityQueue

+ PrometheusMetrics *metrics.PrometheusMetrics

+ func (bpq *PriorityQueue) Ack(id uint64) error

+ func (bpq *PriorityQueue) Create(queueType, queueName string) error

+ func (bpq *PriorityQueue) DeleteQueue() error

+ func (bpq *PriorityQueue) Dequeue(ack bool) (*Message, error)

+ func (bpq *PriorityQueue) Enqueue(id uint64, group string, priority int64, content string, ...) (*Message, error)

+ func (bpq *PriorityQueue) GetMessagesKey(id uint64) []byte

+ func (bpq *PriorityQueue) GetQueueKey(queueName string) []byte

+ func (bpq *PriorityQueue) GetStats() *QueueInfo

+ func (bpq *PriorityQueue) Init(queueType, queueName string) error

+ func (bpq *PriorityQueue) Load(queueName string, loadMessages bool) error

+ func (bpq *PriorityQueue) Nack(id uint64, priority int64, metadata map[string]string) error

+ func (bpq *PriorityQueue) PersistSnapshot(sink raft.SnapshotSink) error

+ func (bpq *PriorityQueue) StartAckQueueMonitoring()

+ func (bpq *PriorityQueue) StopAckQueueMonitoring()

Feb 10, 2025

Changes in this version

type BadgerPriorityQueue

+ func (bpq *BadgerPriorityQueue) PersistSnapshot(sink raft.SnapshotSink) error

type Message

+ QueueName string

+ QueueType string

type QueueManager

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

Jan 8, 2025

Changes in this version

type BadgerPriorityQueue

+ func (bpq *BadgerPriorityQueue) DeleteQueue() error

+ func (bpq *BadgerPriorityQueue) Get(id uint64) (*Message, error)

type DelayedPriorityQueue

+ func (pq *DelayedPriorityQueue) Delete(group string, id uint64) *Item

+ func (pq *DelayedPriorityQueue) Get(group string, id uint64) *Item

type FairPriorityQueue

+ func (fq *FairPriorityQueue) Delete(group string, id uint64) *Item

+ func (fq *FairPriorityQueue) Get(group string, id uint64) *Item

type Message

+ Metadata map[string]string

type PriorityQueue

+ func (pq *PriorityQueue) Delete(id uint64) *Item

+ func (pq *PriorityQueue) Get(id uint64) *Item

type Queue

+ Delete func(group string, id uint64) *Item

+ Get func(group string, id uint64) *Item

type QueueManager

+ func (qm *QueueManager) CreateQueue(queueType, queueName string) (*BadgerPriorityQueue, error)

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

Oct 9, 2024

Changes in this version

type BadgerPriorityQueue

+ func (bpq *BadgerPriorityQueue) Nack(id uint64) error

type QueueStats

+ NackRPS float64

Oct 2, 2024

Changes in this version

+ func NewQueueStats(windowSize int) *queueStats

type BadgerPriorityQueue

+ PrometheusMetrics *PrometheusMetrics

+ func (bpq *BadgerPriorityQueue) GetStats() *QueueInfo

+ type PrometheusMetrics struct

+ func NewPrometheusMetrics(registry prometheus.Registerer, subsystem string) *PrometheusMetrics

+ type QueueInfo struct

+ Name string

+ Ready int64

+ Stats *QueueStats

+ Total int64

+ Type string

+ Unacked int64

type QueueManager

+ PrometheusMetrics *PrometheusMetrics

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

+ func (qm *QueueManager) LoadQueues()

+ func (qm *QueueManager) SetPrometheusMetrics(metrics *PrometheusMetrics)

+ type QueueStats struct

+ AckRPS float64

+ DequeueRPS float64

+ EnqueueRPS float64

Sep 10, 2024

Changes in this version

type BadgerPriorityQueue

+ func (bpq *BadgerPriorityQueue) StartAckQueueMonitoring()

+ func (bpq *BadgerPriorityQueue) StopAckQueueMonitoring()

Sep 5, 2024

Changes in this version

+ var ErrEmptyQueue = fmt.Errorf("queue is empty")

+ var ErrMessageNotFound = fmt.Errorf("message not found")

+ var ErrQueueNotFound = fmt.Errorf("queue not found")

+ type BadgerPriorityQueue struct

+ func NewBadgerPriorityQueue(db *badger.DB) *BadgerPriorityQueue

+ func (bpq *BadgerPriorityQueue) Ack(id uint64) error

+ func (bpq *BadgerPriorityQueue) Create(queueType, queueName string) error

+ func (bpq *BadgerPriorityQueue) Delete() error

+ func (bpq *BadgerPriorityQueue) Dequeue(ack bool) (*Message, error)

+ func (bpq *BadgerPriorityQueue) Enqueue(group string, priority int64, content string) (*Message, error)

+ func (bpq *BadgerPriorityQueue) GetByID(id uint64) (*Message, error)

+ func (bpq *BadgerPriorityQueue) GetMessagesKey(id uint64) []byte

+ func (bpq *BadgerPriorityQueue) GetNextID() (uint64, error)

+ func (bpq *BadgerPriorityQueue) GetQueueKey(queueName string) []byte

+ func (bpq *BadgerPriorityQueue) Init(queueType, queueName string) error

+ func (bpq *BadgerPriorityQueue) Len() int

+ func (bpq *BadgerPriorityQueue) Load(queueName string, loadMessages bool) error

+ func (bpq *BadgerPriorityQueue) UpdatePriority(id uint64, newPriority int64) error

+ type DelayedPriorityQueue struct

+ func NewDelayedPriorityQueue(minFirst bool) *DelayedPriorityQueue

+ func (pq *DelayedPriorityQueue) DeleteByID(group string, id uint64) *Item

+ func (pq *DelayedPriorityQueue) Dequeue() *Item

+ func (pq *DelayedPriorityQueue) Enqueue(group string, item *Item)

+ func (pq *DelayedPriorityQueue) GetByID(group string, id uint64) *Item

+ func (pq *DelayedPriorityQueue) Len() uint64

+ func (pq *DelayedPriorityQueue) UpdatePriority(group string, id uint64, priority int64)

+ type FairPriorityQueue struct

+ func NewFairPriorityQueue() *FairPriorityQueue

+ func (fq *FairPriorityQueue) DeleteByID(group string, id uint64) *Item

+ func (fq *FairPriorityQueue) Dequeue() *Item

+ func (fq *FairPriorityQueue) Enqueue(group string, item *Item)

+ func (fq *FairPriorityQueue) GetByID(group string, id uint64) *Item

+ func (fq *FairPriorityQueue) Len() uint64

+ func (fq *FairPriorityQueue) UpdatePriority(group string, id uint64, priority int64)

+ type Item struct

+ Group int64

+ ID uint64

+ Priority int64

+ func (m *Item) UpdatePriority(newPriority int64)

+ type LinkedList struct

+ func (l *LinkedList) Append(node *LinkedListNode)

+ func (l *LinkedList) Len() uint64

+ func (l *LinkedList) Remove(node *LinkedListNode)

+ type LinkedListNode struct

+ func NewLinkedListNode(group string, queue *PriorityQueue) *LinkedListNode

+ func (n *LinkedListNode) Queue() *PriorityQueue

+ type Message struct

+ Content string

+ Group string

+ ID uint64

+ Priority int64

+ func MessageFromBytes(data []byte) (*Message, error)

+ func (m *Message) ToBytes() ([]byte, error)

+ func (m *Message) UpdatePriority(newPriority int64)

+ type PriorityQueue struct

+ func NewPriorityQueue(minFirst bool) *PriorityQueue

+ func (pq *PriorityQueue) DeleteByID(id uint64) *Item

+ func (pq *PriorityQueue) GetByID(id uint64) *Item

+ func (pq *PriorityQueue) Peek() any

+ func (pq *PriorityQueue) Pop() any

+ func (pq *PriorityQueue) Push(x any)

+ func (pq *PriorityQueue) UpdatePriority(id uint64, priority int64)

+ func (pq PriorityQueue) Len() int

+ func (pq PriorityQueue) Less(i, j int) bool

+ func (pq PriorityQueue) Swap(i, j int)

+ type Queue interface

+ DeleteByID func(group string, id uint64) *Item

+ Dequeue func() *Item

+ Enqueue func(group string, item *Item)

+ GetByID func(group string, id uint64) *Item

+ Len func() uint64

+ UpdatePriority func(group string, id uint64, priority int64)

+ type QueueConfig struct

+ Name string

+ Type string

+ func QueueConfigFromBytes(data []byte) (*QueueConfig, error)

+ func (qc *QueueConfig) ToBytes() ([]byte, error)

+ type QueueManager struct

+ func NewQueueManager(db *badger.DB) *QueueManager

+ func (qm *QueueManager) Create(queueType, queueName string) (*BadgerPriorityQueue, error)

+ func (qm *QueueManager) Delete(queueName string) error

+ func (qm *QueueManager) GetQueue(queueName string) (*BadgerPriorityQueue, error)

+ func (qm *QueueManager) RunValueLogGC()