feat(pipe): add a pipe_passthrough audio backend that writes the raw encoded stream

README.md

@@ -240,15 +240,15 @@ log_disable_timestamp: false # Whether to disable timestamps in log output
 device_id: '' # Spotify device ID (auto-generated)
 device_name: '' # Spotify device name
 device_type: computer # Spotify device type (icon)
-audio_backend: alsa # Audio backend to use (alsa, pipe, pulseaudio)
+audio_backend: alsa # Audio backend to use (alsa, pipe, pipe_passthrough, pulseaudio)
 audio_backend_runtime_socket: '' # Audio backends' runtime socket to use, if backend is pulseaudio
 audio_device: default # ALSA audio device to use for playback
 mixer_device: '' # ALSA mixer device for volume synchronization 
 mixer_control_name: Master # ALSA mixer control name for volume synchronization
 audio_buffer_time: 500000 # Audio buffer time in microseconds, ALSA only
 audio_period_count: 4 # Number of periods to request, ALSA only
-audio_output_pipe: '' # Path to a named pipe for audio output
-audio_output_pipe_format: s16le # Audio output pipe format (s16le, s32le, f32le)
+audio_output_pipe: '' # Path to a named pipe for audio output (pipe and pipe_passthrough backends)
+audio_output_pipe_format: s16le # Audio output pipe format (s16le, s32le, f32le), ignored by pipe_passthrough
 bitrate: 160 # Playback bitrate (96, 160, 320)
 crossfade_duration: 0 # Crossfade duration between tracks in milliseconds (0 to disable)
 volume_steps: 100 # Volume steps count

config_schema.json

@@ -71,7 +71,8 @@
           "enum": [
             "alsa",
             "pulseaudio",
-            "pipe"
+            "pipe",
+            "pipe_passthrough"
           ],
           "default": "alsa"
         },
@@ -107,7 +108,7 @@
         },
         "audio_output_pipe": {
           "type": "string",
-          "description": "The path to an existing FIFO for the pipe audio backend",
+          "description": "The path to an existing FIFO for the pipe and pipe_passthrough audio backends",
           "default": ""
         },
         "audio_output_pipe_format": {

daemon/controls.go

@@ -523,7 +523,18 @@ func (p *AppPlayer) play(ctx context.Context) error {
 	seekPos := p.state.trackPosition()
 	seekPos = max(0, min(seekPos, int64(p.primaryStream.Media.Duration())))
 	if err := p.player.SeekMs(seekPos); err != nil {
-		return fmt.Errorf("failed seeking before play: %w", err)
+		// A passthrough source can only restart, never jump mid-stream. When
+		// the stored position cannot be honored, starting playback where the
+		// stream actually is beats not starting at all: failing here left
+		// the track loaded but never playing, and the player then moved past
+		// it, so a single Next press audibly skipped two tracks. The
+		// position report stays truthful either way, because a passthrough
+		// source derives PositionMs from the bytes actually consumed.
+		if errors.Is(err, player.ErrPassthroughCannotSeek) {
+			p.app.log.WithError(err).Warnf("cannot seek before play, starting from the stream's current position")
+		} else {
+			return fmt.Errorf("failed seeking before play: %w", err)
+		}
 	}
 
 	if err := p.player.Play(); err != nil {

output.go

@@ -16,3 +16,16 @@ type AudioSource interface {
 	// Read reads 32bit little endian floats from the stream
 	Read([]float32) (int, error)
 }
+
+// AudioSourcePassthrough is an AudioSource that can additionally hand out the
+// raw encoded (Ogg/Vorbis) bytes instead of decoded float32 samples. It backs
+// the pipe_passthrough backend: the decoder is bypassed and the container is
+// written to the pipe untouched, so a downstream consumer (e.g. a hardware
+// decoder) does the decoding. ReadBytes and Read must not be mixed on the
+// same source.
+type AudioSourcePassthrough interface {
+	AudioSource
+
+	// ReadBytes reads the raw encoded stream until EOF.
+	ReadBytes([]byte) (int, error)
+}

output/driver-pipe-passthrough.go

@@ -0,0 +1,185 @@
+package output
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"syscall"
+
+	librespot "github.com/devgianlu/go-librespot"
+)
+
+// pipePassthroughOutput implements the pipe_passthrough backend: it writes
+// the raw encoded (Ogg/Vorbis) stream from an AudioSourcePassthrough reader
+// straight to a named pipe, so a downstream consumer with its own decoder
+// (a hardware decoder, another player process, a transcoder) does the
+// decoding.
+//
+// Unlike the pipe backend this driver never sees decoded samples, so that
+// driver's PCM concerns (sample format transform, clamping, volume scaling)
+// do not exist here and audio_output_pipe_format does not apply.
+type pipePassthroughOutput struct {
+	reader librespot.AudioSourcePassthrough
+	file   *os.File
+
+	lock sync.Mutex
+	cond *sync.Cond
+
+	paused bool
+	closed bool
+
+	volumeUpdate chan float32
+	err          chan error
+}
+
+func newPipePassthroughOutput(opts *NewOutputOptions) (*pipePassthroughOutput, error) {
+	if len(opts.OutputPipe) == 0 {
+		return nil, fmt.Errorf("audio_output_pipe must be set for the %s backend", BackendPipePassthrough)
+	}
+
+	reader, ok := opts.Reader.(librespot.AudioSourcePassthrough)
+	if !ok {
+		return nil, fmt.Errorf("passthrough requires an AudioSourcePassthrough reader")
+	}
+
+	out := &pipePassthroughOutput{
+		reader:       reader,
+		err:          make(chan error, 2),
+		volumeUpdate: opts.VolumeUpdate,
+	}
+	out.cond = sync.NewCond(&out.lock)
+
+	// The FIFO is opened exactly like in the pipe driver; the few lines are
+	// duplicated on purpose to keep the two drivers independent.
+	//
+	// Open the FIFO for writing as non-blocking to cause an error if there is no reader.
+	var err error
+	out.file, err = os.OpenFile(opts.OutputPipe, os.O_WRONLY|syscall.O_NONBLOCK, 0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open fifo: %w", err)
+	}
+
+	// Restore blocking mode now that we are sure we have a reader.
+	if err := syscall.SetNonblock(int(out.file.Fd()), false); err != nil {
+		return nil, fmt.Errorf("failed to set blocking mode on fifo: %w", err)
+	}
+
+	go out.outputLoop()
+
+	return out, nil
+}
+
+func (out *pipePassthroughOutput) outputLoop() {
+	buf := make([]byte, 16*1024)
+
+	for {
+		out.lock.Lock()
+
+		for out.paused && !out.closed {
+			out.cond.Wait()
+		}
+
+		if out.closed {
+			out.lock.Unlock()
+			break
+		}
+
+		n, err := out.reader.ReadBytes(buf)
+
+		if n > 0 {
+			if _, werr := out.file.Write(buf[:n]); werr != nil {
+				out.err <- werr
+				out.closed = true
+				out.lock.Unlock()
+				break
+			}
+		}
+
+		if errors.Is(err, io.EOF) {
+			// Reached EOF, move to a "paused" state.
+			out.paused = true
+		} else if err != nil {
+			// Got some other error. Close the output and report the error.
+			out.err <- err
+			out.closed = true
+			out.lock.Unlock()
+			break
+		}
+
+		out.lock.Unlock()
+	}
+
+	_ = out.Close()
+}
+
+func (out *pipePassthroughOutput) Pause() error {
+	out.lock.Lock()
+	defer out.lock.Unlock()
+
+	if out.closed {
+		return nil
+	}
+
+	out.paused = true
+	out.cond.Signal()
+	return nil
+}
+
+func (out *pipePassthroughOutput) Resume() error {
+	out.lock.Lock()
+	defer out.lock.Unlock()
+
+	if out.closed {
+		return nil
+	}
+
+	out.paused = false
+	out.cond.Signal()
+	return nil
+}
+
+// Drop is a no-op: the driver buffers nothing itself, bytes are handed to
+// the pipe as soon as they are read.
+func (out *pipePassthroughOutput) Drop() error {
+	return nil
+}
+
+// DelayMs reports no delay: the encoded stream is opaque to this driver, so
+// it cannot know how much audio the downstream decoder has buffered.
+func (out *pipePassthroughOutput) DelayMs() (int64, error) {
+	return 0, nil
+}
+
+// SetVolume only reports the volume back on the update channel: volume
+// scaling is an operation on decoded samples, the encoded stream is written
+// untouched and the downstream consumer is in charge of volume.
+func (out *pipePassthroughOutput) SetVolume(vol float32) {
+	if vol < 0 || vol > 1 {
+		panic(fmt.Sprintf("invalid volume value: %0.2f", vol))
+	}
+
+	sendVolumeUpdate(out.volumeUpdate, vol)
+}
+
+func (out *pipePassthroughOutput) Error() <-chan error {
+	// No need to lock here (out.err is only set in newPipePassthroughOutput).
+	return out.err
+}
+
+func (out *pipePassthroughOutput) Close() error {
+	out.lock.Lock()
+	defer out.lock.Unlock()
+
+	if out.closed {
+		return nil
+	}
+
+	_ = out.file.Close()
+
+	out.closed = true
+	out.cond.Signal()
+
+	return nil
+}

output/output.go

@@ -6,6 +6,12 @@ import (
 	librespot "github.com/devgianlu/go-librespot"
 )
 
+// BackendPipePassthrough is the audio backend that writes the raw encoded
+// (Ogg/Vorbis) stream to a named pipe instead of decoded PCM. It is served
+// by its own driver, separate from the pipe backend; the Reader must
+// implement AudioSourcePassthrough and OutputPipeFormat is ignored.
+const BackendPipePassthrough = "pipe_passthrough"
+
 type Output interface {
 	// Pause pauses the output.
 	Pause() error
@@ -135,6 +141,12 @@ func NewOutput(options *NewOutputOptions) (Output, error) {
 			return nil, err
 		}
 		return out, nil
+	case BackendPipePassthrough:
+		out, err := newPipePassthroughOutput(options)
+		if err != nil {
+			return nil, err
+		}
+		return out, nil
 	case "audio-toolbox":
 		out, err := newAudioToolboxOutput(options)
 		if err != nil {

player/passthrough_seek_test.go

@@ -0,0 +1,78 @@
+package player
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"testing"
+)
+
+// newTestPassthrough builds a passthrough source over sz bytes representing
+// durMs of audio, with already consumed bytes so PositionMs reports a
+// mid-stream position.
+func newTestPassthrough(t *testing.T, sz, durMs, consumed int64) *passthroughSource {
+	t.Helper()
+	p := newPassthroughSource(bytes.NewReader(make([]byte, sz)), sz, durMs)
+	if consumed > 0 {
+		if _, err := io.CopyN(io.Discard, readerFrom(p), consumed); err != nil {
+			t.Fatalf("consuming %d bytes: %v", consumed, err)
+		}
+	}
+	return p
+}
+
+// readerFrom adapts ReadBytes to io.Reader for the test helper.
+func readerFrom(p *passthroughSource) io.Reader { return passthroughReader{p} }
+
+type passthroughReader struct{ p *passthroughSource }
+
+func (r passthroughReader) Read(b []byte) (int, error) { return r.p.ReadBytes(b) }
+
+// The daemon re-asserts the state's position before every play; that state
+// advances with the wall clock, so the request routinely lands a few
+// milliseconds past the source's actual position. It must be a no-op, not a
+// failure: refusing it failed play() outright, the track never started, and
+// the player moved past it (one Next press audibly skipped two tracks).
+func TestPassthroughSeekToCurrentPositionIsNoOp(t *testing.T) {
+	// Fresh stream at position 0: the classic "seek to 1ms" from a stored
+	// resume position.
+	p := newTestPassthrough(t, 4_000_000, 200_000, 0)
+	if err := p.SetPositionMs(1); err != nil {
+		t.Fatalf("seek to 1ms on a fresh stream must be a no-op, got %v", err)
+	}
+	if got := p.PositionMs(); got != 0 {
+		t.Fatalf("position after no-op seek = %dms, want 0", got)
+	}
+
+	// Mid-stream, re-asserting the current position (resume after pause).
+	p = newTestPassthrough(t, 4_000_000, 200_000, 2_000_000) // ~100s in
+	cur := p.PositionMs()
+	if err := p.SetPositionMs(cur + 40); err != nil {
+		t.Fatalf("re-asserting the current position must succeed, got %v", err)
+	}
+}
+
+// A genuine jump still fails loudly (the controller must snap back to the
+// real position instead of believing a jump the audio never performed), and
+// the error is the exported sentinel so callers can tell it apart.
+func TestPassthroughSeekMidStreamJumpFails(t *testing.T) {
+	p := newTestPassthrough(t, 4_000_000, 200_000, 0)
+	err := p.SetPositionMs(60_000)
+	if err == nil {
+		t.Fatal("mid-stream jump must fail")
+	}
+	if !errors.Is(err, ErrPassthroughCannotSeek) {
+		t.Fatalf("error = %v, want ErrPassthroughCannotSeek", err)
+	}
+}
+
+// Seeking to <= 0 rewinds to the start (the documented restart escape hatch).
+func TestPassthroughSeekZeroRestarts(t *testing.T) {
+	p := newTestPassthrough(t, 4_000_000, 200_000, 1_000_000)
+	if err := p.SetPositionMs(0); err != nil {
+		t.Fatalf("restart seek: %v", err)
+	}
+	if got := p.PositionMs(); got != 0 {
+		t.Fatalf("position after restart = %dms, want 0", got)
+	}
+}