initial structure

queue/queue.go

@@ -0,0 +1,61 @@
+package queue
+
+import (
+	"container/heap"
+)
+
+// An Item is something we manage in a priority queue.
+type Item struct {
+	value    string // id of the item in the full db
+	priority int64  // timestamp of the item in the queue.
+	// The index is needed by update and is maintained by the heap.Interface methods.
+	index int // the index of the item in the heap.
+}
+
+// A PriorityQueue implements heap.Interface and holds Items.
+type PriorityQueue []*Item
+
+func (pq PriorityQueue) Len() int { return len(pq) }
+
+func (pq PriorityQueue) Less(i, j int) bool {
+	// We want Pop to give us the lowest, not highest, priority so we use smaller than here.
+	return pq[i].priority < pq[j].priority
+}
+
+func (pq PriorityQueue) Swap(i, j int) {
+	pq[i], pq[j] = pq[j], pq[i]
+	pq[i].index = i
+	pq[j].index = j
+}
+
+func (pq *PriorityQueue) Push(x interface{}) {
+	n := len(*pq)
+	item := x.(*Item)
+	item.index = n
+	*pq = append(*pq, item)
+}
+
+func (pq *PriorityQueue) Pop() interface{} {
+	old := *pq
+	n := len(old)
+	item := old[n-1]
+	item.index = -1 // for safety
+	*pq = old[0 : n-1]
+	return item
+}
+
+func (pq *PriorityQueue) Look() *Item {
+	if len(*pq) == 0 {
+		return nil
+	}
+	old := *pq
+	item := old[0]
+	return item
+}
+
+// update modifies the priority and value of an Item in the queue.
+func (pq *PriorityQueue) update(item *Item, value string, priority int64) {
+	item.value = value
+	item.priority = priority
+	heap.Fix(pq, item.index)
+}

queue/queue_test.go

@@ -0,0 +1,49 @@
+package queue
+
+import (
+	"container/heap"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQueue(t *testing.T) {
+	// Some items and their priorities.
+	items := map[string]int64{
+		"id0": 2,
+		"id1": 5,
+		"id2": 4,
+	}
+
+	// Create a priority queue, put the items in it, and
+	// establish the priority queue (heap) invariants.
+	pq := make(PriorityQueue, len(items))
+	i := 0
+	for value, priority := range items {
+		pq[i] = &Item{
+			value:    value,
+			priority: priority,
+			index:    i,
+		}
+		i++
+	}
+	heap.Init(&pq)
+
+	assert.Equal(t, pq.Look().value, "id0")
+
+	// Insert a new item and then modify its priority.
+	item := &Item{
+		value:    "id3",
+		priority: 3,
+	}
+	heap.Push(&pq, item)
+	pq.update(item, item.value, 1)
+	assert.Equal(t, pq.Look().value, "id3")
+
+	// Take the items out; they arrive in increasing priority order.
+	for pq.Len() > 0 {
+		_ = heap.Pop(&pq).(*Item)
+		// fmt.Println("priority", item.priority, ":", item.value)
+	}
+	assert.Nil(t, pq.Look())
+}