Merge pull request #2 from fleetspace/priority-queue-delete-item-fix

Fix Peek to not put the head past the tail

priority_queue.go

@@ -238,7 +238,6 @@ func (pq *PriorityQueue) DequeueByID(priority uint8, id uint64) (*PriorityItem,
 	}
 
 	return item, nil
-
 }
 
 // DequeueItems dequeue's a slice of items inside a transaction so either
@@ -254,7 +253,9 @@ func (pq *PriorityQueue) DequeueItems(items []*PriorityItem) error {
 			return err
 		}
 	}
-	return trans.Commit()
+	ret := trans.Commit()
+	_, _ = pq.Peek() // update the head
+	return ret
 }
 
 // Peek returns the next item in the priority queue without removing it.
@@ -538,7 +539,7 @@ func (pq *PriorityQueue) getNextItem() (*PriorityItem, error) {
 			return result, nil
 		}
 		// If we've iterated the whole way through... we have to give up
-		if pq.levels[pq.curLevel].head > pq.levels[pq.curLevel].tail {
+		if pq.levels[pq.curLevel].head >= pq.levels[pq.curLevel].tail {
 			return result, err
 		}
 		// This item was deleted, move on to the next one!

priority_queue_test.go

@@ -1163,6 +1163,53 @@ func TestDequeueItems(t *testing.T) {
 	}
 }
 
+func TestDequeueItemsAll(t *testing.T) {
+	type Vector struct {
+		Data string
+	}
+	tmpdir, err := ioutil.TempDir("", "testdb")
+	if err != nil {
+		t.Error(err)
+	}
+
+	defer os.RemoveAll(tmpdir)
+
+	queue, err := OpenPriorityQueue(tmpdir, ASC)
+	if err != nil {
+		t.Error(err)
+	}
+
+	item1 := Vector{Data: "1"}
+	item2 := Vector{Data: "2"}
+	item3 := Vector{Data: "3"}
+
+	saved1, err := queue.EnqueueObject(1, item1)
+	if err != nil {
+		t.Error(err)
+	}
+	saved2, err := queue.EnqueueObject(1, item2)
+	if err != nil {
+		t.Error(err)
+	}
+	saved3, err := queue.EnqueueObject(1, item3)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if queue.Length() != 3 {
+		t.Errorf("Expected queue length of 3, got %d", queue.Length())
+	}
+
+	err = queue.DequeueItems([]*PriorityItem{saved1, saved2, saved3})
+	if err != nil {
+		t.Error(err)
+	}
+
+	if queue.Length() != 0 {
+		t.Errorf("Expected queue length of 0, got %d", queue.Length())
+	}
+}
+
 func TestDequeueItemsEmpty(t *testing.T) {
 	type Vector struct {
 		Data string