package player import ( "runtime" "testing" "time" "github.com/gopxl/beep/v2" ) // slowStreamer wraps a beep.Streamer and introduces a delay before // each Stream call, simulating slow disk I/O. type slowStreamer struct { inner beep.Streamer delay time.Duration } func (s *slowStreamer) Stream(samples [][2]float64) (int, bool) { time.Sleep(s.delay) return s.inner.Stream(samples) } func (s *slowStreamer) Err() error { return s.inner.Err() } // finiteStreamer produces exactly N samples with incrementing values // starting at 1.0 (so sample 0 → 1.0, sample 1 → 2.0, etc.) and // then signals end-of-stream. Values start at 1 so they are // distinguishable from silence (zero). func finiteStreamer(n int) beep.Streamer { pos := 0 return beep.StreamerFunc(func(samples [][2]float64) (int, bool) { if pos >= n { return 0, false } filled := 0 for i := range samples { if pos >= n { break } val := float64(pos + 1) // +1 so first sample is 1.0 samples[i] = [2]float64{val, val} pos++ filled++ } return filled, true }) } func TestBufferedStreamer_BasicStream(t *testing.T) { const total = 1000 src := finiteStreamer(total) bs := NewBufferedStreamer(src, 2048) defer bs.Close() var collected [][2]float64 buf := make([][2]float64, 256) for { n, ok := bs.Stream(buf) for i := range n { // Skip silence frames the streamer injects whenever its // ring buffer momentarily underruns (not just leading // silence). Real samples always start at 1.0, so any zero // frame is injected silence, never source data. if buf[i][0] == 0 && buf[i][1] == 0 { continue } collected = append(collected, buf[i]) } if !ok { break } // Safety valve: if we've collected enough samples plus // extra from potential silence padding, break. if len(collected) >= total { // Drain remaining. for { n, ok = bs.Stream(buf) if !ok { break } for i := range n { if buf[i][0] != 0 || buf[i][1] != 0 { collected = append(collected, buf[i]) } } } break } } if len(collected) != total { t.Fatalf( "expected %d samples, got %d", total, len(collected), ) } // Verify ordering (values start at 1.0). for i, s := range collected { expected := float64(i + 1) if s[0] != expected || s[1] != expected { t.Fatalf( "sample %d: expected [%f %f], got [%f %f]", i, expected, expected, s[0], s[1], ) } } } func TestBufferedStreamer_SmallReads(t *testing.T) { const total = 200 src := finiteStreamer(total) bs := NewBufferedStreamer(src, 512) defer bs.Close() // Give read-ahead time to fill. time.Sleep(50 * time.Millisecond) var collected [][2]float64 buf := make([][2]float64, 1) // Read one sample at a time. for { n, ok := bs.Stream(buf) for i := range n { if buf[i][0] == 0 && buf[i][1] == 0 && len(collected) == 0 { continue } collected = append(collected, buf[i]) } if !ok { break } if len(collected) >= total { // Drain. for { n, ok = bs.Stream(buf) if !ok { break } for i := range n { if buf[i][0] != 0 || buf[i][1] != 0 { collected = append(collected, buf[i]) } } } break } } if len(collected) != total { t.Fatalf( "expected %d samples, got %d", total, len(collected), ) } for i, s := range collected { expected := float64(i + 1) if s[0] != expected || s[1] != expected { t.Fatalf( "sample %d: expected [%f %f], got [%f %f]", i, expected, expected, s[0], s[1], ) } } } func TestBufferedStreamer_SourceDrained(t *testing.T) { const total = 100 src := finiteStreamer(total) bs := NewBufferedStreamer(src, 256) defer bs.Close() // Wait for read-ahead to completely drain the source. time.Sleep(50 * time.Millisecond) // Read all samples out. consumed := 0 buf := make([][2]float64, 32) hitEOF := false for range 1000 { // Safety limit. n, ok := bs.Stream(buf) for i := range n { if buf[i][0] != 0 || buf[i][1] != 0 { consumed++ } } if !ok { hitEOF = true break } } if !hitEOF { t.Fatal("expected stream to return ok=false after source drained") } if consumed != total { t.Fatalf("expected %d non-zero samples, got %d", total, consumed) } } func TestBufferedStreamer_EmptyBufferReturnsSilence(t *testing.T) { // Use a slow source that sleeps 50ms per call. src := &slowStreamer{ inner: finiteStreamer(100), delay: 50 * time.Millisecond, } bs := NewBufferedStreamer(src, 1024) defer bs.Close() // Immediately call Stream before read-ahead has had time to // fill anything. The buffer should be empty. buf := make([][2]float64, 64) n, ok := bs.Stream(buf) if !ok { t.Fatal("expected ok=true when buffer is empty but source not drained") } if n != len(buf) { t.Fatalf("expected %d samples (silence), got %d", len(buf), n) } // All returned samples should be silence (zeros). for i := range n { if buf[i][0] != 0 || buf[i][1] != 0 { t.Fatalf( "sample %d should be silence, got [%f %f]", i, buf[i][0], buf[i][1], ) } } } func TestBufferedStreamer_Flush(t *testing.T) { // Use a slow source so the read-ahead goroutine cannot fully // drain it before we flush. Each chunk sleeps 5ms, giving us // time to flush while data is still being produced. src := &slowStreamer{ inner: finiteStreamer(5000), delay: 5 * time.Millisecond, } bs := NewBufferedStreamer(src, 2048) defer bs.Close() // Let read-ahead fill some data. time.Sleep(50 * time.Millisecond) // Read a few samples to confirm data is buffered. buf := make([][2]float64, 32) n, ok := bs.Stream(buf) if !ok || n == 0 { t.Fatal("expected buffered data before flush") } // Record the last sample value we saw. lastBefore := buf[n-1][0] // Verify the buffer had more data than we consumed (i.e. // there's stale data in the ring that Flush should discard). bs.mu.Lock() countBeforeFlush := bs.count bs.mu.Unlock() if countBeforeFlush == 0 { t.Fatal("expected non-empty ring buffer before flush") } // Flush discards all buffered data. bs.Flush() // Verify the ring buffer is empty after flush. bs.mu.Lock() countAfterFlush := bs.count bs.mu.Unlock() if countAfterFlush != 0 { t.Fatalf( "expected 0 samples after flush, got %d", countAfterFlush, ) } // Wait for read-ahead to refill with fresh data. time.Sleep(100 * time.Millisecond) // The next non-zero sample must come from AFTER the // pre-flush position in the source (i.e. its value must be // greater than lastBefore + countBeforeFlush, since those // samples were discarded). foundNonZero := false for range 200 { n, ok = bs.Stream(buf) for i := range n { if buf[i][0] != 0 { // The sample must be strictly greater than what // was buffered before flush. if buf[i][0] <= lastBefore { t.Fatalf( "after flush, got sample value %f "+ "which is <= pre-flush value %f "+ "(stale data not discarded)", buf[i][0], lastBefore, ) } foundNonZero = true break } } if foundNonZero || !ok { break } time.Sleep(1 * time.Millisecond) } if !foundNonZero { t.Fatal("expected non-zero samples after flush") } } func TestBufferedStreamer_Close(t *testing.T) { // Use a source that never drains. infinite := beep.StreamerFunc(func(samples [][2]float64) (int, bool) { for i := range samples { samples[i] = [2]float64{1.0, 1.0} } return len(samples), true }) goroutinesBefore := runtime.NumGoroutine() bs := NewBufferedStreamer(infinite, 4096) // Let read-ahead goroutine start. time.Sleep(10 * time.Millisecond) bs.Close() // Wait for goroutine to exit. time.Sleep(50 * time.Millisecond) goroutinesAfter := runtime.NumGoroutine() // The goroutine count should not have increased. Allow ±1 for // runtime fluctuations. if goroutinesAfter > goroutinesBefore+1 { t.Fatalf( "goroutine leak: before=%d after=%d", goroutinesBefore, goroutinesAfter, ) } // Calling Close again should not panic. bs.Close() }