queue_go/queue.go

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
}