Mixer v2 + fix for memory leak

compositors.go

@@ -230,39 +230,10 @@ func Seq(s ...Streamer) Streamer {
 //
 // Mix does not propagate errors from the Streamers.
 func Mix(s ...Streamer) Streamer {
-	return StreamerFunc(func(samples [][2]float64) (n int, ok bool) {
-		var tmp [512][2]float64
-
-		for len(samples) > 0 {
-			toStream := min(len(tmp), len(samples))
-
-			// clear the samples
-			clear(samples[:toStream])
-
-			snMax := 0 // max number of streamed samples in this iteration
-			for _, st := range s {
-				// mix the stream
-				sn, sok := st.Stream(tmp[:toStream])
-				if sn > snMax {
-					snMax = sn
-				}
-				ok = ok || sok
-
-				for i := range tmp[:sn] {
-					samples[i][0] += tmp[i][0]
-					samples[i][1] += tmp[i][1]
-				}
-			}
-
-			n += snMax
-			if snMax < len(tmp) {
-				break
-			}
-			samples = samples[snMax:]
-		}
-
-		return n, ok
-	})
+	return &Mixer{
+		streamers:     s,
+		stopWhenEmpty: true,
+	}
 }
 
 // Dup returns two Streamers which both stream the same data as the original s. The two Streamers

compositors_test.go

@@ -209,9 +209,7 @@ func TestMix(t *testing.T) {
 
 	got := testtools.Collect(beep.Mix(s...))
 
-	if !reflect.DeepEqual(want, got) {
-		t.Error("Mix not working correctly")
-	}
+	testtools.AssertSamplesEqual(t, want, got)
 }
 
 func TestDup(t *testing.T) {

mixer.go

@@ -1,9 +1,17 @@
 package beep
 
 // Mixer allows for dynamic mixing of arbitrary number of Streamers. Mixer automatically removes
-// drained Streamers. Mixer's stream never drains, when empty, Mixer streams silence.
+// drained Streamers. Depending on the KeepAlive() setting, Stream will either play silence or
+// drain when all Streamers have been drained. By default, Mixer keeps playing silence.
 type Mixer struct {
-	streamers []Streamer
+	streamers     []Streamer
+	stopWhenEmpty bool
+}
+
+// KeepAlive configures the Mixer to either keep playing silence when all its Streamers have
+// drained (keepAlive == true) or stop playing (keepAlive == false).
+func (m *Mixer) KeepAlive(keepAlive bool) {
+	m.stopWhenEmpty = !keepAlive
 }
 
 // Len returns the number of Streamers currently playing in the Mixer.
@@ -18,12 +26,20 @@ func (m *Mixer) Add(s ...Streamer) {
 
 // Clear removes all Streamers from the mixer.
 func (m *Mixer) Clear() {
+	for i := range m.streamers {
+		m.streamers[i] = nil
+	}
 	m.streamers = m.streamers[:0]
 }
 
-// Stream streams all Streamers currently in the Mixer mixed together. This method always returns
-// len(samples), true. If there are no Streamers available, this methods streams silence.
+// Stream the samples of all Streamers currently in the Mixer mixed together. Depending on the
+// KeepAlive() setting, Stream will either play silence or drain when all Streamers have been
+// drained.
 func (m *Mixer) Stream(samples [][2]float64) (n int, ok bool) {
+	if m.stopWhenEmpty && len(m.streamers) == 0 {
+		return 0, false
+	}
+
 	var tmp [512][2]float64
 
 	for len(samples) > 0 {
@@ -32,19 +48,29 @@ func (m *Mixer) Stream(samples [][2]float64) (n int, ok bool) {
 		// clear the samples
 		clear(samples[:toStream])
 
+		snMax := 0
 		for si := 0; si < len(m.streamers); si++ {
 			// mix the stream
 			sn, sok := m.streamers[si].Stream(tmp[:toStream])
 			for i := range tmp[:sn] {
 				samples[i][0] += tmp[i][0]
 				samples[i][1] += tmp[i][1]
 			}
-			if !sok {
+			if sn > snMax {
+				snMax = sn
+			}
+
+			if sn < toStream || !sok {
 				// remove drained streamer
-				sj := len(m.streamers) - 1
-				m.streamers[si], m.streamers[sj] = m.streamers[sj], m.streamers[si]
-				m.streamers = m.streamers[:sj]
+				last := len(m.streamers) - 1
+				m.streamers[si] = m.streamers[last]
+				m.streamers[last] = nil
+				m.streamers = m.streamers[:last]
 				si--
+
+				if m.stopWhenEmpty && len(m.streamers) == 0 {
+					return n + snMax, true
+				}
 			}
 		}
 

mixer_test.go

@@ -51,7 +51,7 @@ func TestMixer_MixesSamples(t *testing.T) {
 
 func TestMixer_DrainedStreamersAreRemoved(t *testing.T) {
 	s1, _ := testtools.RandomDataStreamer(50)
-	s2, _ := testtools.RandomDataStreamer(60)
+	s2, _ := testtools.RandomDataStreamer(65)
 
 	m := beep.Mixer{}
 	m.Add(s1)
@@ -62,13 +62,12 @@ func TestMixer_DrainedStreamersAreRemoved(t *testing.T) {
 	assert.Len(t, samples, 50)
 	assert.Equal(t, 2, m.Len())
 
-	// Fully drain s1.
-	// Drain s2 but not so far it returns false.
+	// Drain s1 (s1 returns !ok, n == 0)
 	samples = testtools.CollectNum(10, &m)
 	assert.Len(t, samples, 10)
 	assert.Equal(t, 1, m.Len())
 
-	// Fully drain s2.
+	// Drain s2 (s2 returns ok, n < len(samples))
 	samples = testtools.CollectNum(10, &m)
 	assert.Len(t, samples, 10)
 	assert.Equal(t, 0, m.Len())
@@ -82,22 +81,16 @@ func TestMixer_PlaysSilenceWhenNoStreamersProduceSamples(t *testing.T) {
 	assert.Len(t, samples, 10)
 	assert.Equal(t, make([][2]float64, 10), samples)
 
-	// Test silence after streamer is partly drained.
-	s, _ := testtools.RandomDataStreamer(50)
+	// Test silence after streamer has only streamed part of the requested samples.
+	s, data := testtools.RandomDataStreamer(50)
 	m.Add(s)
-
 	samples = testtools.CollectNum(100, &m)
 	assert.Len(t, samples, 100)
-	assert.Equal(t, 1, m.Len())
-	assert.Equal(t, make([][2]float64, 50), samples[50:])
-
-	// Test silence when streamer is fully drained.
-	samples = testtools.CollectNum(10, &m)
-	assert.Len(t, samples, 10)
 	assert.Equal(t, 0, m.Len())
-	assert.Equal(t, make([][2]float64, 10), samples)
+	assert.Equal(t, data, samples[:50])
+	assert.Equal(t, make([][2]float64, 50), samples[50:])
 
-	// Test silence after streamer was fully drained.
+	// Test silence after streamers have been drained & removed.
 	samples = testtools.CollectNum(10, &m)
 	assert.Len(t, samples, 10)
 	assert.Equal(t, make([][2]float64, 10), samples)