github-featured/bjarneo--cliamp
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(player): prevent stuck radio ffmpeg streams

Browse Source
This commit is contained in:
Bjarne Øverli
2026-06-13 16:11:17 +02:00
parent 213ada6919
commit 52d7fce824
5 changed files with 137 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
player/decode.go
+2 -1
View File
@@ -30,6 +30,7 @@ var SupportedExts = map[string]bool{
".ogg": true,
".m4a": true,
".aac": true,
".aacp": true,
".m4b": true,
".alac": true,
".wma": true,
@@ -226,7 +227,7 @@ func formatExt(path string) string {
// needsFFmpeg reports whether the given extension requires ffmpeg to decode.
func needsFFmpeg(ext string) bool {
switch ext {
case ".m4a", ".aac", ".m4b", ".alac", ".wma", ".opus", ".webm":
case ".m4a", ".aac", ".aacp", ".m4b", ".alac", ".wma", ".opus", ".webm":
return true
}
return false
player/decode_test.go
+25
View File
@@ -12,3 +12,28 @@ func TestIsHLS(t *testing.T) {
}
}
}
func TestNeedsFFmpeg(t *testing.T) {
tests := []struct {
ext string
want bool
}{
{ext: ".aac", want: true},
{ext: ".aacp", want: true},
{ext: ".opus", want: true},
{ext: ".mp3", want: false},
{ext: ".ogg", want: false},
}
for _, tt := range tests {
if got := needsFFmpeg(tt.ext); got != tt.want {
t.Errorf("needsFFmpeg(%q) = %v, want %v", tt.ext, got, tt.want)
}
}
}
func TestSupportedExtsIncludesAACP(t *testing.T) {
if !SupportedExts[".aacp"] {
t.Fatal("SupportedExts[.aacp] = false, want true")
}
}
player/ffmpeg.go
+45 -7
View File
@@ -12,6 +12,7 @@ import (
"path/filepath"
"strconv"
"strings"
"time"
"github.com/gopxl/beep/v2"
)
@@ -22,6 +23,9 @@ const pcmFrameSize16 = 4
// pcmFrameSize32 is the byte size of one stereo f32le sample frame (2 channels × 4 bytes).
const pcmFrameSize32 = 8
// ffmpegPipeTimeout limits how long URL streams may take to produce initial PCM.
const ffmpegPipeTimeout = 15 * time.Second
// pcmFrameSize returns the byte size of one stereo sample frame for the given format.
func pcmFrameSize(f32 bool) int {
if f32 {
@@ -231,15 +235,22 @@ func (f *ffmpegPipe) Position() int { return f.pos }
// 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()
src := f.src
pipe := f.pipe
cmd := f.cmd
f.src = nil
f.pipe = nil
f.cmd = nil
if src != nil {
src.Close()
}
if f.pipe != nil {
f.pipe.Close()
if pipe != nil {
pipe.Close()
}
if f.cmd != nil && f.cmd.Process != nil {
f.cmd.Process.Kill()
f.cmd.Wait()
if cmd != nil && cmd.Process != nil {
cmd.Process.Kill()
cmd.Wait()
}
}
@@ -255,6 +266,33 @@ func (f *ffmpegPipe) bitDepth() int {
return 16
}
// waitForInitialAudio waits until ffmpeg has produced at least one PCM byte.
// This runs before a URL stream is handed to the speaker, so an idle or broken
// live stream cannot park the audio goroutine in Read and block future swaps.
func (f *ffmpegPipe) waitForInitialAudio(timeout time.Duration) error {
peekErr := make(chan error, 1)
go func() {
_, err := f.reader.Peek(1)
peekErr <- err
}()
timer := time.NewTimer(timeout)
defer timer.Stop()
select {
case err := <-peekErr:
if err != nil {
f.stop()
return fmt.Errorf("waiting for audio data: %w", err)
}
return nil
case <-timer.C:
f.stop()
<-peekErr // drain after stop unblocks the pipe reader
return fmt.Errorf("timed out waiting for audio data (%v)", timeout)
}
}
// ffmpegPipeStreamer reads PCM data incrementally from a running ffmpeg process.
// Used for live/infinite streams where seeking is not supported.
type ffmpegPipeStreamer struct {
player/icy_test.go
+35
View File
@@ -4,6 +4,7 @@ import (
"bytes"
"io"
"os/exec"
"strings"
"testing"
"time"
@@ -129,3 +130,37 @@ func TestFFmpegPipeStreamCloseUnblocks(t *testing.T) {
t.Fatal("Close() hung: stdin-copy goroutine was not unblocked")
}
}
func TestFFmpegPipeStreamInitialAudioTimeoutCloses(t *testing.T) {
if _, err := exec.LookPath("ffmpeg"); err != nil {
t.Skip("ffmpeg not installed")
}
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)
}
err = dec.waitForInitialAudio(100 * time.Millisecond)
if err == nil {
t.Fatal("waitForInitialAudio returned nil, want timeout")
}
if !strings.Contains(err.Error(), "timed out waiting for audio data") {
t.Fatalf("waitForInitialAudio error = %v, want timeout", err)
}
done := make(chan struct{})
go func() {
dec.Close()
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("Close() hung after initial audio timeout")
}
}
player/pipeline.go
+30 -4
View File
@@ -87,6 +87,17 @@ func (p *Player) buildPipeline(path string) (*trackPipeline, error) {
return p.buildPipelineAt(path, 0, 0)
}
func (p *Player) decodeFFmpegURLStream(path string) (*ffmpegPipeStreamer, beep.Format, error) {
decoder, format, err := decodeFFmpegStream(path, p.sr, p.bitDepth)
if err != nil {
return nil, beep.Format{}, err
}
if err := decoder.waitForInitialAudio(ffmpegPipeTimeout); err != nil {
return nil, beep.Format{}, err
}
return decoder, format, nil
}
// buildPipelineAt is like buildPipeline but starts the HTTP stream at byteOffset
// (using a Range: bytes=N- header) and records timeOffset as the playback origin.
// For local files byteOffset is ignored; use decoder.Seek instead.
@@ -165,7 +176,7 @@ func (p *Player) buildPipelineAt(path string, byteOffset int64, timeOffset time.
// window. Feeding the playlist bytes via stdin (the needsFFmpeg path below)
// would strip the base URL and break relative segment resolution.
if isURL(path) && isHLS(ext) {
decoder, format, err := decodeFFmpegStream(path, p.sr, p.bitDepth)
decoder, format, err := p.decodeFFmpegURLStream(path)
if err != nil {
return nil, fmt.Errorf("open hls: %w", err)
}
@@ -204,7 +215,7 @@ func (p *Player) buildPipelineAt(path string, byteOffset int64, timeOffset time.
tp, err := p.buildChainedOggPipeline(rc, onMeta)
if err != nil {
rc.Close()
decoder, fmt2, err2 := decodeFFmpegStream(path, p.sr, p.bitDepth)
decoder, fmt2, err2 := p.decodeFFmpegURLStream(path)
if err2 != nil {
return nil, fmt.Errorf("decode: %w", err2)
}
@@ -231,6 +242,9 @@ func (p *Player) buildPipelineAt(path string, byteOffset int64, timeOffset time.
rc.Close()
return nil, fmt.Errorf("decode: %w", err)
}
if err := decoder.waitForInitialAudio(ffmpegPipeTimeout); err != nil {
return nil, fmt.Errorf("decode: %w", err)
}
return &trackPipeline{
decoder: decoder,
stream: decoder,
@@ -274,8 +288,20 @@ func (p *Player) buildPipelineAt(path string, byteOffset int64, timeOffset time.
if needsFFmpeg(ext) {
return nil, fmt.Errorf("decode: %w", err)
}
// Native decoder failed (e.g., IEEE float WAV). Fall back to ffmpeg,
// which reads from the path directly and handles more formats.
if isURL(path) {
decoder, format, err := p.decodeFFmpegURLStream(path)
if err != nil {
return nil, fmt.Errorf("decode: %w", err)
}
return &trackPipeline{
decoder: decoder,
stream: decoder,
format: format,
path: path,
}, nil
}
// Native local decoder failed (e.g., IEEE float WAV). Fall back to
// buffered ffmpeg decode, which handles more formats.
decoder, format, err = decodeFFmpeg(path, p.sr, p.bitDepth)
if err != nil {
return nil, fmt.Errorf("decode: %w", err)