fix(player): keep ICY metadata reader attached for ffmpeg radio streams

HTTP streams that need ffmpeg (AAC, AAC+, Opus, ...) closed the ICY-wrapped reader and handed ffmpeg the URL directly, so ffmpeg opened its own connection and StreamTitle never reached the player. Live radio on these codecs showed no track info even when the server sent perfect ICY metadata. Feed ffmpeg from the existing reader chain via stdin (pipe:0) instead of the URL, keeping the icyReader in the data path. Extract the shared ffmpeg-pipe launch into startFFmpegPipe and fold the optional stdin source into ffmpegPipe so stop() closes it before cmd.Wait() (the os/exec stdin-copy goroutine parks in src.Read on an infinite stream and would otherwise deadlock). Adds tests for parseStreamTitle, the icyReader metaint boundary handling, and the stdin-fed streamer's clean shutdown.
2026-05-29 18:42:28 +02:00
parent 7024e87c5f
commit 21bbc747d0
3 changed files with 187 additions and 17 deletions
@@ -162,10 +162,22 @@ func decodeFFmpegStream(path string, sr beep.SampleRate, bitDepth int) (*ffmpegP
 		ext := filepath.Ext(path)
 		return nil, beep.Format{}, fmt.Errorf("ffmpeg is required to play %s files — install it with your package manager", ext)
 	}
+	fp, format, err := startFFmpegPipe(path, nil, sr, bitDepth)
+	if err != nil {
+		return nil, beep.Format{}, err
+	}
+	return &ffmpegPipeStreamer{ffmpegPipe: fp}, format, nil
+}

+// startFFmpegPipe launches ffmpeg transcoding input to raw PCM on stdout and
+// returns an ffmpegPipe reading that stdout. input is the ffmpeg -i argument
+// (a URL/path, or "pipe:0" when feeding via stdin); stdin, when non-nil, is
+// wired to the process. Callers add the concrete Seek behavior by embedding the
+// returned ffmpegPipe in a streamer type.
+func startFFmpegPipe(input string, stdin io.ReadCloser, sr beep.SampleRate, bitDepth int) (ffmpegPipe, beep.Format, error) {
 	pcmFmt, codec, precision := ffmpegPCMArgs(bitDepth)
 	cmd := exec.Command("ffmpeg",
-		"-i", path,
+		"-i", input,
 		"-f", pcmFmt,
 		"-acodec", codec,
 		"-ar", strconv.Itoa(int(sr)),
@@ -173,22 +185,19 @@ func decodeFFmpegStream(path string, sr beep.SampleRate, bitDepth int) (*ffmpegP
 		"-loglevel", "error",
 		"pipe:1",
 	)
+	cmd.Stdin = stdin

 	pipe, err := cmd.StdoutPipe()
 	if err != nil {
-		return nil, beep.Format{}, fmt.Errorf("ffmpeg stdout pipe: %w", err)
+		return ffmpegPipe{}, beep.Format{}, fmt.Errorf("ffmpeg stdout pipe: %w", err)
 	}
 	if err := cmd.Start(); err != nil {
-		return nil, beep.Format{}, fmt.Errorf("ffmpeg start: %w", err)
+		return ffmpegPipe{}, beep.Format{}, fmt.Errorf("ffmpeg start: %w", err)
 	}

-	format := beep.Format{
-		SampleRate:  sr,
-		NumChannels: 2,
-		Precision:   precision,
-	}
-
-	return &ffmpegPipeStreamer{ffmpegPipe: ffmpegPipe{cmd: cmd, reader: bufio.NewReaderSize(pipe, pipeBufSize), pipe: pipe, f32: bitDepth == 32}}, format, nil
+	fp := ffmpegPipe{cmd: cmd, reader: bufio.NewReaderSize(pipe, pipeBufSize), pipe: pipe, src: stdin, f32: bitDepth == 32}
+	format := beep.Format{SampleRate: sr, NumChannels: 2, Precision: precision}
+	return fp, format, nil
 }

 // ffmpegPipe holds the common state and methods shared by all pipe-based
@@ -198,6 +207,7 @@ type ffmpegPipe struct {
 	cmd    *exec.Cmd
 	reader *bufio.Reader
 	pipe   io.ReadCloser
+	src    io.ReadCloser        // optional stdin source (e.g. ICY-wrapped body); closed on stop
 	buf    [pcmFrameSize32]byte // large enough for both 16-bit and 32-bit frames
 	f32    bool                 // true = f32le, false = s16le
 	err    error
@@ -215,8 +225,15 @@ func (f *ffmpegPipe) Err() error    { return f.err }
 func (f *ffmpegPipe) Len() int      { return f.total }
 func (f *ffmpegPipe) Position() int { return f.pos }

-// stop kills the running ffmpeg process and cleans up.
+// stop kills the running ffmpeg process and cleans up. When stdin is fed from
+// src, src is closed first: os/exec runs a goroutine copying src -> ffmpeg
+// stdin, and cmd.Wait() blocks until it returns. For an infinite radio stream
+// that goroutine is parked in src.Read, so src must be closed to unblock it
+// before Wait, otherwise stop hangs.
 func (f *ffmpegPipe) stop() {
+	if f.src != nil {
+		f.src.Close()
+	}
 	if f.pipe != nil {
 		f.pipe.Close()
 	}
@@ -246,6 +263,23 @@ type ffmpegPipeStreamer struct {

 func (f *ffmpegPipeStreamer) Seek(int) error { return nil }

+// decodeFFmpegPipeStream starts ffmpeg reading from src via stdin (pipe:0)
+// instead of letting ffmpeg open the URL itself. Keeping the caller's reader
+// chain in the data path means the ICY metadata reader stays attached, so live
+// radio StreamTitle parsing keeps working for ffmpeg-only codecs (AAC, AAC+,
+// Opus, ...). src is closed when the stream stops. Used for live/infinite HTTP
+// streams; seeking is not supported.
+func decodeFFmpegPipeStream(src io.ReadCloser, sr beep.SampleRate, bitDepth int) (*ffmpegPipeStreamer, beep.Format, error) {
+	if _, err := exec.LookPath("ffmpeg"); err != nil {
+		return nil, beep.Format{}, fmt.Errorf("ffmpeg is required to play this stream — install it with your package manager")
+	}
+	fp, format, err := startFFmpegPipe("pipe:0", src, sr, bitDepth)
+	if err != nil {
+		return nil, beep.Format{}, err
+	}
+	return &ffmpegPipeStreamer{ffmpegPipe: fp}, format, nil
+}
+
 // decodeFFmpegLocal starts ffmpeg as a streaming pipe for local files, giving
 // instant playback start instead of buffering the entire file to memory.
 // Seeking is supported by killing and restarting ffmpeg with a -ss offset.
@@ -0,0 +1,131 @@
+package player
+
+import (
+	"bytes"
+	"io"
+	"os/exec"
+	"testing"
+	"time"
+
+	"github.com/gopxl/beep/v2"
+)
+
+func TestParseStreamTitle(t *testing.T) {
+	tests := []struct {
+		name string
+		meta string
+		want string
+	}{
+		{"artist and title", "StreamTitle='Daft Punk - Aerodynamic';StreamUrl='';", "Daft Punk - Aerodynamic"},
+		{"title only", "StreamTitle='Some Show';", "Some Show"},
+		{"empty title", "StreamTitle='';StreamUrl='';", ""},
+		{"no stream title key", "StreamUrl='https://example.com';", ""},
+		{"empty block", "", ""},
+		{"missing trailing semicolon", "StreamTitle='No Semicolon'", "No Semicolon"},
+		{"title containing semicolon", "StreamTitle='A; B - C';StreamUrl='';", "A; B - C"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := parseStreamTitle(tt.meta); got != tt.want {
+				t.Errorf("parseStreamTitle(%q) = %q, want %q", tt.meta, got, tt.want)
+			}
+		})
+	}
+}
+
+// icyBlock encodes one metadata block: a 1-byte length prefix (size/16) followed
+// by the null-padded metadata, matching the SHOUTcast/Icecast wire format.
+func icyBlock(meta string) []byte {
+	if meta == "" {
+		return []byte{0}
+	}
+	n := (len(meta) + 15) / 16
+	out := make([]byte, 1+n*16)
+	out[0] = byte(n)
+	copy(out[1:], meta)
+	return out
+}
+
+func TestIcyReaderStripsMetadataAndReportsTitles(t *testing.T) {
+	const metaInt = 8
+	var raw bytes.Buffer
+	raw.WriteString("AAAAAAAA") // audio block 1
+	raw.Write(icyBlock("StreamTitle='Song One';"))
+	raw.WriteString("BBBBBBBB") // audio block 2
+	raw.Write(icyBlock(""))     // empty metadata block (no change)
+	raw.WriteString("CCCCCCCC") // audio block 3
+	raw.Write(icyBlock("StreamTitle='Song Two';"))
+	raw.WriteString("DDDDDDDD") // audio block 4 (partial, no trailing meta)
+
+	var titles []string
+	r := newIcyReader(io.NopCloser(&raw), metaInt, func(s string) {
+		titles = append(titles, s)
+	})
+
+	// Read in small chunks to exercise the metaint-boundary clamping.
+	got, err := io.ReadAll(&chunkedReader{r: r, n: 3})
+	if err != nil {
+		t.Fatalf("ReadAll: %v", err)
+	}
+
+	const wantAudio = "AAAAAAAABBBBBBBBCCCCCCCCDDDDDDDD"
+	if string(got) != wantAudio {
+		t.Errorf("audio = %q, want %q", got, wantAudio)
+	}
+
+	wantTitles := []string{"Song One", "Song Two"}
+	if len(titles) != len(wantTitles) {
+		t.Fatalf("titles = %v, want %v", titles, wantTitles)
+	}
+	for i, w := range wantTitles {
+		if titles[i] != w {
+			t.Errorf("titles[%d] = %q, want %q", i, titles[i], w)
+		}
+	}
+}
+
+// chunkedReader caps each Read at n bytes so tests can drive a reader with many
+// small reads, exercising boundary handling.
+type chunkedReader struct {
+	r io.Reader
+	n int
+}
+
+func (c *chunkedReader) Read(p []byte) (int, error) {
+	if len(p) > c.n {
+		p = p[:c.n]
+	}
+	return c.r.Read(p)
+}
+
+// TestFFmpegPipeStreamCloseUnblocks verifies that closing the stdin-fed ffmpeg
+// streamer does not hang when its source is a live stream parked in Read. os/exec
+// copies src -> ffmpeg stdin in a goroutine that cmd.Wait() joins, so Close must
+// close src first to unblock it. Regression guard for the AAC ICY fix.
+func TestFFmpegPipeStreamCloseUnblocks(t *testing.T) {
+	if _, err := exec.LookPath("ffmpeg"); err != nil {
+		t.Skip("ffmpeg not installed")
+	}
+
+	// io.Pipe with no writer simulates an open-but-idle radio body: Read blocks.
+	pr, pw := io.Pipe()
+	t.Cleanup(func() { pw.Close() })
+
+	dec, _, err := decodeFFmpegPipeStream(pr, beep.SampleRate(44100), 16)
+	if err != nil {
+		t.Fatalf("decodeFFmpegPipeStream: %v", err)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		dec.Close()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Close() hung: stdin-copy goroutine was not unblocked")
+	}
+}
@@ -203,17 +203,22 @@ func (p *Player) buildPipelineAt(path string, byteOffset int64, timeOffset time.

 	// For HTTP streams that need ffmpeg (e.g. AAC+), use the streaming
 	// pipe decoder so playback starts immediately instead of buffering
-	// the entire (potentially infinite) stream.
+	// the entire (potentially infinite) stream. Feed ffmpeg from the existing
+	// reader chain via stdin rather than handing it the URL: this keeps the
+	// ICY metadata reader attached so live radio StreamTitle parsing works for
+	// ffmpeg-only codecs (AAC, AAC+, Opus, ...).
 	if isURL(path) && needsFFmpeg(ext) {
-		rc.Close()
-		decoder, format, err := decodeFFmpegStream(path, p.sr, p.bitDepth)
+		decoder, format, err := decodeFFmpegPipeStream(rc, p.sr, p.bitDepth)
 		if err != nil {
+			rc.Close()
 			return nil, fmt.Errorf("decode: %w", err)
 		}
 		return &trackPipeline{
-			decoder: decoder,
-			stream:  decoder,
-			format:  format,
+			decoder:       decoder,
+			stream:        decoder,
+			format:        format,
+			bytesRead:     byteCounter,
+			contentLength: src.contentLength,
 		}, nil
 	}