diff --git a/queue.go b/queue.go index 1b35667..7261754 100644 --- a/queue.go +++ b/queue.go @@ -2,7 +2,6 @@ package queue import ( "container/heap" - "loafle.com/overflow/agent_api/observer" "log" "sync" "time" @@ -16,11 +15,9 @@ type Item struct { type LoafleQueue struct { mtx sync.Mutex items []*Item - queueType string interval time.Duration size int - eventChanel chan interface{} senderChanel chan interface{} } @@ -86,36 +83,34 @@ func (lq LoafleQueue) newItem(value interface{}) *Item { } } -func (lq *LoafleQueue) notifyEventHandler(c chan interface{}) { - for data := range c { - it := lq.newItem(data) - heap.Push(lq, it) - } +func (lq *LoafleQueue) PushItem(v interface{}) { + + it := lq.newItem(v) + heap.Push(lq, it) + } func (lq *LoafleQueue) Close() { - if lq.eventChanel != nil { - observer.Remove(lq.queueType, lq.eventChanel) - } + //if lq.eventChanel != nil { + // observer.Remove(lq.queueType, lq.eventChanel) + //} lq.items = nil } -func NewQueue(eventType string, interval time.Duration, senderChanel chan interface{}) *LoafleQueue { + +func NewQueue(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) + //observer.Add(eventType, lq.eventChanel) + //go lq.notifyEventHandler(event) go lq.getItems() return lq