queue_go/queue.go

package queue

import (
	"container/heap"
	"loafle.com/overflow/agent_api/observer"
	"log"
	"time"
)

type EventType int

const (
	DATA_TYPE  EventType = 0
	EVENT_TYPE EventType = 1
)

type Item struct {
	Value    interface{}
	Priority int
}

type LoafleQueue struct {
	items     []*Item
	queueType string
	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) StringType(tp EventType) string {
	var tempType string

	if tp == DATA_TYPE {
		tempType = "DATA_TYPE"
	} else if tp == EVENT_TYPE {
		tempType = "EVENT_TYPE"
	} else {
		log.Fatal("Event Type Error")
	}

	return tempType
}

func (lq *LoafleQueue) Close() {
	if lq.eventChanel != nil {
		observer.Remove(lq.queueType, lq.eventChanel)
	}
}
func NewQueue(eventType EventType, interval time.Duration) *LoafleQueue {
	items := make([]*Item, 0)
	event := make(chan interface{}, 0)

	lq := &LoafleQueue{
		items: items,
		//queueType:   tempType,
		interval:    interval,
		eventChanel: event,
	}

	lq.queueType = lq.StringType(eventType)

	heap.Init(lq)

	observer.Add(lq.queueType, lq.eventChanel)
	go lq.notifyEventHandler(event)

	return lq
}