Fix off-by-one error in Queue.ForEach()

queue.go

@@ -1,6 +1,8 @@
 package msgbus
 
 import (
+	"fmt"
+
 	sync "github.com/sasha-s/go-deadlock"
 )
 
@@ -113,7 +115,7 @@ func (q *Queue) Peek() interface{} {
 }
 
 // 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.
+// access into the queue in O(n) time.
 func (q *Queue) ForEach(f func(elem interface{}) error) error {
 	q.Lock()
 	defer q.Unlock()
@@ -122,8 +124,13 @@ func (q *Queue) ForEach(f func(elem interface{}) error) error {
 		return nil
 	}
 
-	for i := 0; i < q.count; i++ {
-		if err := f(q.buf[i]); err != nil {
+	fmt.Printf("buf: #%v\n", q.buf)
+	fmt.Printf("head: %d\n", q.head)
+	fmt.Printf("tail: %d\n", q.tail)
+	fmt.Printf("count: %d\n", q.count)
+
+	for i := 0; i < (q.count - 1); i++ {
+		if err := f(q.buf[q.next(i)]); err != nil {
 			return err
 		}
 	}

queue_test.go

@@ -55,6 +55,39 @@ func TestForEach(t *testing.T) {
 	assert.Equal(ys, xs)
 }
 
+func TestForEachOverflow(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+
+	maxlen := 4
+	q := &Queue{maxlen: maxlen}
+
+	N := (maxlen + 1)
+	var ys []int
+	for i := 0; i < N; i++ {
+		q.Push(i)
+		ys = append(ys, i)
+		t.Logf("pushed %d", i)
+		t.Logf("q.Len() %d", q.Len())
+		t.Logf("q.Size() %d", q.Size())
+	}
+	assert.False(q.Full())
+	assert.Equal(N, q.Len())
+
+	var xs []int
+	err := q.ForEach(func(e interface{}) error {
+		i, ok := e.(int)
+		if !ok {
+			return fmt.Errorf("unexpected type %T", e)
+		}
+		xs = append(xs, i)
+		return nil
+	})
+	require.NoError(err)
+	t.Logf("#%v == #%v\n", ys[(N-maxlen):], xs)
+	assert.Equal(ys[(N-maxlen):], xs)
+}
+
 func TestMaxLen(t *testing.T) {
 	q := Queue{maxlen: minCapacity}
 	assert.Equal(t, q.MaxLen(), minCapacity)