Fix queue tests

queue.go

@@ -2,7 +2,6 @@ package msgbus
 
 import (
 	sync "github.com/sasha-s/go-deadlock"
-	log "github.com/sirupsen/logrus"
 )
 
 // minCapacity is the smallest capacity that queue may have.
@@ -62,23 +61,6 @@ func (q *Queue) Empty() bool {
 	return q.count == 0
 }
 
-// ForEach applys the function `f` over each item in the queue for read-only
-// access into the queue in O(n) time for indexining into the queue.
-func (q *Queue) ForEach(f func(elem interface{}) error) error {
-	q.Lock()
-	defer q.Unlock()
-
-	log.Debugf("buf: #%v", q.buf)
-	log.Debugf("q.tail: %d", q.tail)
-	log.Debugf("q.head: %d", q.head)
-	for i := q.head; i < q.tail; i++ {
-		if err := f(q.buf[i]); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 // Full returns true if the queue is full false otherwise
 func (q *Queue) Full() bool {
 	q.RLock()
@@ -130,6 +112,25 @@ func (q *Queue) Peek() interface{} {
 	return q.buf[q.head]
 }
 
+// ForEach applys the function `f` over each item in the queue for read-only
+// access into the queue in O(n) time for indexining into the queue.
+func (q *Queue) ForEach(f func(elem interface{}) error) error {
+	q.Lock()
+	defer q.Unlock()
+
+	if q.count <= 0 {
+		return nil
+	}
+
+	for i := 0; i < q.count; i++ {
+		if err := f(q.buf[i]); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // next returns the next buffer position wrapping around buffer.
 func (q *Queue) next(i int) int {
 	return (i + 1) & (len(q.buf) - 1) // bitwise modulus
@@ -141,7 +142,7 @@ func (q *Queue) growIfFull() {
 		q.buf = make([]interface{}, minCapacity)
 		return
 	}
-	if q.count == len(q.buf) && q.count < q.maxlen {
+	if q.count == len(q.buf) && (q.maxlen == 0 || q.count < q.maxlen) {
 		q.resize()
 	}
 }

queue_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestEmpty(t *testing.T) {
-	q := Queue{}
+	q := NewQueue()
 	assert.Zero(t, q.Len())
 	assert.True(t, q.Empty())
 }
@@ -17,11 +17,12 @@ func TestEmpty(t *testing.T) {
 func TestSimple(t *testing.T) {
 	assert := assert.New(t)
 
-	q := Queue{}
+	q := NewQueue()
 
 	for i := 0; i < minCapacity; i++ {
 		q.Push(i)
 	}
+	assert.Equal(minCapacity, q.Len())
 
 	for i := 0; i < minCapacity; i++ {
 		assert.Equal(q.Peek(), i)
@@ -39,7 +40,7 @@ func TestForEach(t *testing.T) {
 	for _, y := range ys {
 		q.Push(y)
 	}
-	assert.Equal(q.Len(), 4)
+	assert.Equal(4, q.Len())
 
 	var xs []int
 	err := q.ForEach(func(e interface{}) error {
@@ -70,7 +71,7 @@ func TestFull(t *testing.T) {
 }
 
 func TestBufferWrap(t *testing.T) {
-	q := Queue{}
+	q := NewQueue()
 
 	for i := 0; i < minCapacity; i++ {
 		q.Push(i)
@@ -90,7 +91,7 @@ func TestBufferWrap(t *testing.T) {
 func TestLen(t *testing.T) {
 	assert := assert.New(t)
 
-	q := Queue{}
+	q := NewQueue()
 	assert.Zero(q.Len())
 
 	for i := 0; i < 1000; i++ {
@@ -105,14 +106,14 @@ func TestLen(t *testing.T) {
 }
 
 func BenchmarkPush(b *testing.B) {
-	q := Queue{}
+	q := NewQueue()
 	for i := 0; i < b.N; i++ {
 		q.Push(i)
 	}
 }
 
 func BenchmarkPushPop(b *testing.B) {
-	q := Queue{}
+	q := NewQueue()
 	for i := 0; i < b.N; i++ {
 		q.Push(i)
 	}