package queue import ( "container/heap" "loafle.com/overflow/agent_api/observer" "log" "sync" "time" ) type Item struct { Value interface{} Priority int } type LoafleQueue struct { mtx sync.Mutex items []*Item queueType string interval time.Duration size int eventChanel chan interface{} senderChanel 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() { resultItems := make([]*Item, 0) for { time.Sleep(time.Second * lq.interval) if lq.Len() > 0 { lq.size = lq.Len() lq.mtx.Lock() for i := 0; i < lq.size; i++ { item := heap.Pop(lq).(*Item) resultItems = append(resultItems, item) } lq.mtx.Unlock() lq.senderChanel <- resultItems lq.size = lq.Len() //log.Println("result length: ", len(resultItems)) resultItems = nil } } } func (lq *LoafleQueue) Push(i interface{}) { lq.mtx.Lock() defer lq.mtx.Unlock() 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, lq.eventChanel) } lq.items = nil } func NewQueue(eventType string, interval time.Duration, senderChanel chan interface{}) *LoafleQueue { items := make([]*Item, 0) event := make(chan interface{}, 0) log.Println("NewQueu Start") lq := &LoafleQueue{ items: items, queueType: eventType, interval: interval, eventChanel: event, senderChanel: senderChanel, } heap.Init(lq) observer.Add(eventType, lq.eventChanel) go lq.notifyEventHandler(event) go lq.getItems() return lq }