96 lines
1.7 KiB
Go
96 lines
1.7 KiB
Go
package queue
|
|
|
|
import (
|
|
"container/heap"
|
|
"loafle.com/overflow/agent_api/observer"
|
|
"time"
|
|
)
|
|
|
|
type Item struct {
|
|
Value interface{}
|
|
Priority int
|
|
}
|
|
|
|
type LoafleQueue struct {
|
|
items []*Item
|
|
queueType observer.KEY
|
|
interval time.Duration
|
|
|
|
eventChanel chan interface{}
|
|
}
|
|
|
|
func (lq LoafleQueue) Len() int {
|
|
return len(lq.items)
|
|
}
|
|
|
|
func (lq LoafleQueue) Less(i, j int) bool {
|
|
|
|
return lq.items[i].Priority < lq.items[j].Priority
|
|
}
|
|
|
|
func (lq LoafleQueue) Swap(i, j int) {
|
|
lq.items[i], lq.items[j] = lq.items[j], lq.items[i]
|
|
}
|
|
|
|
func (lq *LoafleQueue) GetItems() []*Item {
|
|
|
|
time.Sleep(time.Second * lq.interval)
|
|
resultItems := make([]*Item, 0)
|
|
for lq.Len() > 0 {
|
|
item := heap.Pop(lq).(*Item)
|
|
resultItems = append(resultItems, item)
|
|
}
|
|
return resultItems
|
|
}
|
|
|
|
func (lq *LoafleQueue) Push(i interface{}) {
|
|
n := len(lq.items)
|
|
nItem := i.(*Item)
|
|
nItem.Priority = n
|
|
lq.items = append(lq.items, nItem)
|
|
}
|
|
|
|
func (lq *LoafleQueue) Pop() interface{} {
|
|
old := lq.items
|
|
n := len(old)
|
|
nItem := old[n-1]
|
|
lq.items = old[0 : n-1]
|
|
return nItem
|
|
}
|
|
func (lq LoafleQueue) newItem(value interface{}) *Item {
|
|
return &Item{
|
|
Value: value,
|
|
}
|
|
}
|
|
|
|
func (lq *LoafleQueue) notifyEventHandler(c chan interface{}) {
|
|
for data := range c {
|
|
it := lq.newItem(data)
|
|
heap.Push(lq, it)
|
|
}
|
|
}
|
|
|
|
func (lq *LoafleQueue) Close() {
|
|
if lq.eventChanel != nil {
|
|
observer.Remove(lq.queueType.String(), lq.eventChanel)
|
|
}
|
|
}
|
|
func NewQueue(eventType observer.KEY, interval time.Duration) *LoafleQueue {
|
|
items := make([]*Item, 0)
|
|
event := make(chan interface{}, 0)
|
|
|
|
lq := &LoafleQueue{
|
|
items: items,
|
|
queueType: eventType,
|
|
interval: interval,
|
|
eventChanel: event,
|
|
}
|
|
|
|
heap.Init(lq)
|
|
|
|
observer.Add(eventType.String(), lq.eventChanel)
|
|
go lq.notifyEventHandler(event)
|
|
|
|
return lq
|
|
}
|