Test & improve Loop edge cases

- Propagate l.s.Seek() errors.
- Make the loop count of 0 backwards compatible
- Test error handling

compositors.go

@@ -77,10 +77,11 @@ func LoopBetween(start, end int) LoopOption {
 // The returned Streamer propagates any errors from s.
 func Loop(count int, s StreamSeeker, opts ...LoopOption) Streamer {
 	l := &loop{
-		s:       s,
-		remains: count,
-		start:   0,
-		end:     math.MaxInt,
+		s:        s,
+		remains:  count,
+		finished: count == 0,
+		start:    0,
+		end:      math.MaxInt,
 	}
 	for _, opt := range opts {
 		opt(l)
@@ -99,26 +100,29 @@ func Loop(count int, s StreamSeeker, opts ...LoopOption) Streamer {
 }
 
 type loop struct {
-	s       StreamSeeker
-	remains int
-	start   int // start position in the stream where looping begins. Samples before this position are played once before the first loop.
-	end     int // end position in the stream where looping ends and restarts from `start`.
+	s        StreamSeeker
+	remains  int // number of seeks remaining.
+	finished bool
+	start    int // start position in the stream where looping begins. Samples before this position are played once before the first loop.
+	end      int // end position in the stream where looping ends and restarts from `start`.
+	err      error
 }
 
 func (l *loop) Stream(samples [][2]float64) (n int, ok bool) {
-	if l.s.Err() != nil {
+	if l.finished || l.err != nil {
 		return 0, false
 	}
 	for len(samples) > 0 {
 		toStream := len(samples)
 		if l.remains != 0 {
 			samplesUntilEnd := l.end - l.s.Position()
-			if samplesUntilEnd == 0 {
+			if samplesUntilEnd <= 0 {
 				// End of loop, reset the position and decrease the loop count.
 				if l.remains > 0 {
 					l.remains--
 				}
 				if err := l.s.Seek(l.start); err != nil {
+					l.err = err
 					return n, true
 				}
 				continue
@@ -130,6 +134,8 @@ func (l *loop) Stream(samples [][2]float64) (n int, ok bool) {
 		sn, sok := l.s.Stream(samples[:toStream])
 		n += sn
 		if sn < toStream || !sok {
+			l.err = l.s.Err()
+			l.finished = true
 			return n, n > 0
 		}
 		samples = samples[sn:]
@@ -138,7 +144,7 @@ func (l *loop) Stream(samples [][2]float64) (n int, ok bool) {
 }
 
 func (l *loop) Err() error {
-	return l.s.Err()
+	return l.err
 }
 
 // Seq takes zero or more Streamers and returns a Streamer which streams them one by one without pauses.

compositors_test.go

@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/gopxl/beep/v2"
@@ -28,9 +29,10 @@ func TestTake(t *testing.T) {
 
 func TestLoop(t *testing.T) {
 	// Test no loop.
-	s, data := testtools.NewSequentialDataStreamer(5)
+	// For backwards compatibility, a loop count of 0 means that nothing at all will be played.
+	s, _ := testtools.NewSequentialDataStreamer(5)
 	got := testtools.Collect(beep.Loop(0, s))
-	assert.Equal(t, data, got)
+	assert.Empty(t, got)
 
 	// Test loop once.
 	s, _ = testtools.NewSequentialDataStreamer(5)
@@ -66,6 +68,52 @@ func TestLoop(t *testing.T) {
 	s, _ = testtools.NewSequentialDataStreamer(5)
 	got = testtools.CollectNum(10, beep.Loop(-1, s, beep.LoopBetween(2, 4)))
 	assert.Equal(t, [][2]float64{{0, 0}, {1, 1}, {2, 2}, {3, 3}, {2, 2}, {3, 3}, {2, 2}, {3, 3}, {2, 2}, {3, 3}}, got)
+
+	// Test streaming from the middle of the loops.
+	s, _ = testtools.NewSequentialDataStreamer(5)
+	l := beep.Loop(2, s, beep.LoopBetween(2, 4)) // 0, 1, 2, 3, 2, 3, 2, 3
+	// First stream to the middle of a loop.
+	buf := make([][2]float64, 3)
+	if n, ok := l.Stream(buf); n != 3 || !ok {
+		t.Fatalf("want n %d got %d, want ok %t got %t", 5, n, true, ok)
+	}
+	assert.Equal(t, [][2]float64{{0, 0}, {1, 1}, {2, 2}}, buf)
+	// Then stream starting at the middle of the loop.
+	if n, ok := l.Stream(buf); n != 3 || !ok {
+		t.Fatalf("want n %d got %d, want ok %t got %t", 5, n, true, ok)
+	}
+	assert.Equal(t, [][2]float64{{3, 3}, {2, 2}, {3, 3}}, buf)
+
+	// Test error handling in middle of loop.
+	expectedErr := errors.New("expected error")
+	s, _ = testtools.NewSequentialDataStreamer(5)
+	s = testtools.NewDelayedErrorStreamer(s, 5, expectedErr)
+	l = beep.Loop(3, s, beep.LoopBetween(2, 4)) // 0, 1, 2, 3, 2, 3, 2, 3
+	buf = make([][2]float64, 10)
+	if n, ok := l.Stream(buf); n != 5 || !ok {
+		t.Fatalf("want n %d got %d, want ok %t got %t", 5, n, true, ok)
+	}
+	assert.Equal(t, [][2]float64{{0, 0}, {1, 1}, {2, 2}, {3, 3}, {2, 2}, {0, 0}, {0, 0}, {0, 0}, {0, 0}, {0, 0}}, buf)
+	assert.Equal(t, expectedErr, l.Err())
+	if n, ok := l.Stream(buf); n != 0 || ok {
+		t.Fatalf("want n %d got %d, want ok %t got %t", 0, n, false, ok)
+	}
+	assert.Equal(t, expectedErr, l.Err())
+
+	// Test error handling during call to Seek().
+	s, _ = testtools.NewSequentialDataStreamer(5)
+	s = testtools.NewSeekErrorStreamer(s, expectedErr)
+	l = beep.Loop(3, s, beep.LoopBetween(2, 4)) // 0, 1, 2, 3, [error]
+	buf = make([][2]float64, 10)
+	if n, ok := l.Stream(buf); n != 4 || !ok {
+		t.Fatalf("want n %d got %d, want ok %t got %t", 4, n, true, ok)
+	}
+	assert.Equal(t, [][2]float64{{0, 0}, {1, 1}, {2, 2}, {3, 3}, {0, 0}, {0, 0}, {0, 0}, {0, 0}, {0, 0}, {0, 0}}, buf)
+	assert.Equal(t, expectedErr, l.Err())
+	if n, ok := l.Stream(buf); n != 0 || ok {
+		t.Fatalf("want n %d got %d, want ok %t got %t", 0, n, false, ok)
+	}
+	assert.Equal(t, expectedErr, l.Err())
 }
 
 func TestSeq(t *testing.T) {

ctrl_test.go

@@ -48,6 +48,6 @@ func TestCtrl_PropagatesErrors(t *testing.T) {
 	assert.NoError(t, ctrl.Err())
 
 	err := errors.New("oh no")
-	ctrl.Streamer = testtools.ErrorStreamer{Error: err}
+	ctrl.Streamer = testtools.NewErrorStreamer(err)
 	assert.Equal(t, err, ctrl.Err())
 }

internal/testtools/streamers.go

@@ -65,14 +65,101 @@ func (ds *dataStreamer) Seek(p int) error {
 	return nil
 }
 
+// NewErrorStreamer returns a streamer which errors immediately with the given err.
+func NewErrorStreamer(err error) beep.StreamSeeker {
+	return &ErrorStreamer{
+		s: beep.StreamerFunc(func(samples [][2]float64) (n int, ok bool) {
+			panic("unreachable")
+		}),
+		samplesLeft: 0,
+		Error:       err,
+	}
+}
+
+// NewDelayedErrorStreamer wraps streamer s but returns an error after numSamples have been streamed.
+func NewDelayedErrorStreamer(s beep.Streamer, numSamples int, err error) beep.StreamSeeker {
+	return &ErrorStreamer{
+		s:           s,
+		samplesLeft: numSamples,
+		Error:       err,
+	}
+}
+
 type ErrorStreamer struct {
-	Error error
+	s           beep.Streamer
+	samplesLeft int
+	Error       error
+}
+
+func (e *ErrorStreamer) Stream(samples [][2]float64) (n int, ok bool) {
+	if e.samplesLeft == 0 {
+		return 0, false
+	}
+
+	toStream := min(e.samplesLeft, len(samples))
+	n, ok = e.s.Stream(samples[:toStream])
+	e.samplesLeft -= n
+
+	return n, ok
+}
+
+func (e *ErrorStreamer) Err() error {
+	if e.samplesLeft == 0 {
+		return e.Error
+	} else {
+		return e.s.Err()
+	}
+}
+
+func (e *ErrorStreamer) Seek(p int) error {
+	if s, ok := e.s.(beep.StreamSeeker); ok {
+		return s.Seek(p)
+	}
+	panic("source streamer is not a beep.StreamSeeker")
+}
+
+func (e *ErrorStreamer) Len() int {
+	if s, ok := e.s.(beep.StreamSeeker); ok {
+		return s.Len()
+	}
+	panic("source streamer is not a beep.StreamSeeker")
+}
+
+func (e *ErrorStreamer) Position() int {
+	if s, ok := e.s.(beep.StreamSeeker); ok {
+		return s.Position()
+	}
+	panic("source streamer is not a beep.StreamSeeker")
+}
+
+func NewSeekErrorStreamer(s beep.StreamSeeker, err error) *SeekErrorStreamer {
+	return &SeekErrorStreamer{
+		s:   s,
+		err: err,
+	}
+}
+
+type SeekErrorStreamer struct {
+	s   beep.StreamSeeker
+	err error
+}
+
+func (s *SeekErrorStreamer) Stream(samples [][2]float64) (n int, ok bool) {
+	return s.s.Stream(samples)
+}
+
+func (s *SeekErrorStreamer) Err() error {
+	return s.s.Err()
+}
+
+func (s *SeekErrorStreamer) Len() int {
+	return s.s.Len()
 }
 
-func (e ErrorStreamer) Stream(samples [][2]float64) (n int, ok bool) {
-	return 0, false
+func (s *SeekErrorStreamer) Position() int {
+	return s.s.Position()
 }
 
-func (e ErrorStreamer) Err() error {
-	return e.Error
+func (s *SeekErrorStreamer) Seek(p int) error {
+	return s.err
 }