commit d33392d0fd7c0fa984698cd6670b5081a8be247f
parent 4204f85890a6af43cf105af3a3e14df76c12a741
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Fri, 18 Sep 2026 13:15:18 -0400
audio: add a streaming decode API
The only way to decode was into one buffer, and PCM runs roughly an
order of magnitude larger than the compressed input. A few minutes of
stereo audio is tens of megabytes, which rules the package out for long
files and for anything feeding a pipeline incrementally.
Stream and StreamFile return an io.Reader over the same PCM, with the
format known before the first read. Windows decodes incrementally: the
Media Foundation objects move into a Stream that owns and releases them,
where before they were built and torn down inside one call. The ffmpeg
backend pipes straight out of the process rather than buffering, and
reaps its child whether the stream is drained or abandoned. macOS still
decodes up front behind the same interface, so behaviour is uniform and
that backend can improve without another API change.
Abandoning a stream mid-decode no longer strands a locked Media
Foundation buffer, and closing a Decoder now waits for open streams
before releasing the platform state they are still using.
Diffstat:
10 files changed, 638 insertions(+), 36 deletions(-)
diff --git a/README.md b/README.md
@@ -60,13 +60,20 @@ play.File("audio.m4a")
closing one does not disturb another.
- `Decoder.DecodeFile(path)` and `Decoder.Decode(data)` return s16le PCM
and a `Format`. The core package has no audio-output dependency.
+- `Decoder.StreamFile(path)` and `Decoder.Stream(data)` return a `Stream`,
+ an `io.Reader` over the same PCM, so a long track never has to sit in
+ memory whole. `Format` is known before the first read. Close it when
+ done. Windows decodes incrementally and the ffmpeg backend pipes; macOS
+ currently decodes up front and serves from memory.
- `Format` reports `SampleRate`, `Channels` and `BytesPerSample`, which is
always 2.
- A `Decoder` is safe for concurrent use, and `Close` waits for decodes
already in flight.
-- `FFmpegLoad` and `FFmpegDecode` shell out to ffmpeg regardless of
- platform, and need no `Decoder`.
-- The `play` subpackage plays PCM through oto on every platform. Its
+- `FFmpegLoad`, `FFmpegDecode` and `FFmpegStream` shell out to ffmpeg
+ regardless of platform, and need no `Decoder`.
+- The `play` subpackage plays PCM through oto. The core package builds
+ for every target Go supports; `play` is bounded by its audio backend,
+ which needs cgo on Linux and has no FreeBSD support. Its
context is fixed to the first file's sample rate and channel count for
the life of the process.
@@ -77,5 +84,5 @@ Foundation bindings internal.
## TODO
-- [ ] streaming API (current API is a buffered for simplicity)
+- [ ] macOS: decode incrementally rather than buffering behind `Stream`
- [ ] Linux: something better then shelling out to FFmpeg
diff --git a/audio.go b/audio.go
@@ -29,10 +29,12 @@ var ErrClosed = errors.New("nativeaudio: decoder is closed")
// of a program can hold their own without one tearing down the other.
//
// A Decoder is safe for concurrent use. Decodes may run in parallel,
-// and Close waits for those in flight to finish.
+// and Close waits for those in flight, and for any open Stream to be
+// closed, before releasing platform state.
type Decoder struct {
- mu sync.RWMutex
- closed bool
+ mu sync.RWMutex
+ closed bool
+ streams sync.WaitGroup
}
// New creates a Decoder, initialising any platform state the backend
@@ -53,6 +55,9 @@ func (d *Decoder) Close() error {
return nil
}
d.closed = true
+ // Wait for open streams before tearing down platform state, which
+ // their decoder objects are still using.
+ d.streams.Wait()
if err := end(); err != nil {
return fmt.Errorf("shutting down platform decoder: %w", err)
}
@@ -88,3 +93,52 @@ type Format struct {
Channels int // number of channels.
BytesPerSample int // bytes per sample; always 2, for the s16le output this package produces.
}
+
+// Stream decodes compressed audio held in memory, returning PCM through
+// an [io.Reader] rather than a single buffer.
+//
+// The returned Stream must be closed. Only the Windows backend decodes
+// incrementally today; elsewhere the audio is decoded up front and
+// served from memory, which is correct but saves nothing.
+func (d *Decoder) Stream(compressed []byte) (*Stream, error) {
+ d.mu.RLock()
+ defer d.mu.RUnlock()
+ if d.closed {
+ return nil, ErrClosed
+ }
+ s, err := openStream(compressed)
+ if err != nil {
+ return nil, err
+ }
+ d.track(s)
+ return s, nil
+}
+
+// StreamFile decodes the audio file at path, returning PCM through an
+// [io.Reader] rather than a single buffer.
+//
+// The returned Stream must be closed. Backends that shell out to ffmpeg
+// pipe the decode directly, so the PCM is never held whole; the Windows
+// backend reads the compressed file into memory first, which is small
+// next to the PCM it avoids buffering.
+func (d *Decoder) StreamFile(path string) (*Stream, error) {
+ d.mu.RLock()
+ defer d.mu.RUnlock()
+ if d.closed {
+ return nil, ErrClosed
+ }
+ s, err := openStreamFile(path)
+ if err != nil {
+ return nil, err
+ }
+ d.track(s)
+ return s, nil
+}
+
+// track registers an open Stream so Close can wait for it. The caller
+// holds at least a read lock, which is what keeps this from racing the
+// Wait in Close.
+func (d *Decoder) track(s *Stream) {
+ d.streams.Add(1)
+ s.done = d.streams.Done
+}
diff --git a/audio_ffmpeg.go b/audio_ffmpeg.go
@@ -9,11 +9,14 @@ package nativeaudio
import (
"bytes"
"encoding/json"
+ "errors"
"fmt"
+ "io"
"os"
"os/exec"
"strconv"
"strings"
+ "sync"
)
// FFmpegLoad raw PCM with ffmpeg.
@@ -143,3 +146,77 @@ func (s stream) Format() (Format, error) {
BytesPerSample: 2,
}, nil
}
+
+// ffmpegStream pipes PCM out of a running ffmpeg process.
+type ffmpegStream struct {
+ cmd *exec.Cmd
+ stdout io.ReadCloser
+ stderr *bytes.Buffer
+ waitOnce sync.Once
+ waitErr error
+ drained bool
+}
+
+// reap waits for ffmpeg exactly once, whoever gets there first.
+func (f *ffmpegStream) reap() error {
+ f.waitOnce.Do(func() { f.waitErr = f.cmd.Wait() })
+ return f.waitErr
+}
+
+func (f *ffmpegStream) Read(p []byte) (int, error) {
+ n, err := f.stdout.Read(p)
+ if errors.Is(err, io.EOF) {
+ // The pipe closing means ffmpeg is finished, so collect its exit
+ // status: a decode that failed halfway still reaches EOF here.
+ f.drained = true
+ if werr := f.reap(); werr != nil {
+ return n, fmt.Errorf("ffmpeg: %w: %s", werr, f.stderr.String())
+ }
+ }
+ return n, err
+}
+
+func (f *ffmpegStream) Close() error {
+ if !f.drained {
+ // Abandoned early. Kill it rather than leave ffmpeg blocked
+ // writing into a pipe nobody is reading, then reap the corpse.
+ // The resulting wait error is ours, not a decode failure.
+ _ = f.cmd.Process.Kill()
+ _ = f.reap()
+ return nil
+ }
+ return f.reap()
+}
+
+// FFmpegStream decodes an audio file with ffmpeg, returning PCM through
+// an [io.Reader] as ffmpeg produces it.
+//
+// ffmpeg -i <path> -f s16le -
+//
+// Unlike FFmpegLoad this never holds the whole decode in memory. The
+// returned Stream must be closed, including when abandoned early.
+func FFmpegStream(path string) (*Stream, error) {
+ format, err := probe(path)
+ if err != nil {
+ return nil, fmt.Errorf("probing file for metadata: %w", err)
+ }
+ cmd := exec.Command(
+ "ffmpeg",
+ "-i", path,
+ "-f", "s16le",
+ "-",
+ )
+ stdout, err := cmd.StdoutPipe()
+ if err != nil {
+ return nil, fmt.Errorf("attaching to ffmpeg output: %w", err)
+ }
+ stderr := bytes.NewBuffer(nil)
+ cmd.Stderr = stderr
+ if err := cmd.Start(); err != nil {
+ return nil, fmt.Errorf("starting ffmpeg: %w", err)
+ }
+ return &Stream{
+ r: &ffmpegStream{cmd: cmd, stdout: stdout, stderr: stderr},
+ format: format,
+ }, nil
+}
diff --git a/audio_linux.go b/audio_linux.go
@@ -19,3 +19,18 @@ func start() error {
func end() error {
return nil
}
+
+// openStream decodes up front, because ffmpeg cannot read compressed
+// audio from a pipe for every container we support.
+func openStream(by []byte) (*Stream, error) {
+ pcm, format, err := FFmpegDecode(by)
+ if err != nil {
+ return nil, err
+ }
+ return newBufferedStream(pcm, format), nil
+}
+
+// openStreamFile pipes PCM straight out of ffmpeg.
+func openStreamFile(path string) (*Stream, error) {
+ return FFmpegStream(path)
+}
diff --git a/audio_macos.go b/audio_macos.go
@@ -561,3 +561,22 @@ type ErrOSStatus C.OSStatus
func (e ErrOSStatus) Error() string {
return fmt.Sprintf("%v", C.OSStatus(e))
}
+
+// openStream decodes up front and serves the result from memory.
+// AudioToolbox can decode incrementally, but this backend does not yet.
+func openStream(by []byte) (*Stream, error) {
+ pcm, format, err := decode(by)
+ if err != nil {
+ return nil, err
+ }
+ return newBufferedStream(pcm, format), nil
+}
+
+// openStreamFile decodes up front and serves the result from memory.
+func openStreamFile(path string) (*Stream, error) {
+ pcm, format, err := load(path)
+ if err != nil {
+ return nil, err
+ }
+ return newBufferedStream(pcm, format), nil
+}
diff --git a/audio_unknown.go b/audio_unknown.go
@@ -24,3 +24,18 @@ func start() error {
func end() error {
return nil
}
+
+// openStream decodes up front, because ffmpeg cannot read compressed
+// audio from a pipe for every container we support.
+func openStream(by []byte) (*Stream, error) {
+ pcm, format, err := FFmpegDecode(by)
+ if err != nil {
+ return nil, err
+ }
+ return newBufferedStream(pcm, format), nil
+}
+
+// openStreamFile pipes PCM straight out of ffmpeg.
+func openStreamFile(path string) (*Stream, error) {
+ return FFmpegStream(path)
+}
diff --git a/audio_windows.go b/audio_windows.go
@@ -43,3 +43,30 @@ func decode(compressed []byte) (uncompressed []byte, format Format, err error) {
BytesPerSample: f.BytesPerSample,
}, nil
}
+
+// openStream decodes incrementally through Media Foundation.
+func openStream(compressed []byte) (*Stream, error) {
+ ms, err := mf.Open(compressed)
+ if err != nil {
+ return nil, err
+ }
+ f := ms.Format()
+ return &Stream{
+ r: ms,
+ format: Format{
+ SampleRate: f.SampleRate,
+ Channels: f.Channels,
+ BytesPerSample: f.BytesPerSample,
+ },
+ }, nil
+}
+
+// openStreamFile buffers the compressed file, which is small next to
+// the PCM the stream avoids holding, then decodes it incrementally.
+func openStreamFile(path string) (*Stream, error) {
+ by, err := os.ReadFile(path)
+ if err != nil {
+ return nil, fmt.Errorf("reading input file: %w", err)
+ }
+ return openStream(by)
+}
diff --git a/internal/mf/mf.go b/internal/mf/mf.go
@@ -39,57 +39,143 @@ func Shutdown() error {
return MFShutdown()
}
-// Decode compressed audio, returning the uncompressed data as PCM data
-// (s16le) and details about the PCM required to playback correctly.
-func Decode(compressed []byte) (uncompressed []byte, format Format, err error) {
- stream := SHCreateMemStream(unsafe.SliceData(compressed), len(compressed))
- if stream == nil {
- return nil, format, fmt.Errorf("could not allocate IStream")
+// Stream decodes audio incrementally, implementing [io.ReadCloser] over
+// s16le PCM. It owns the Media Foundation objects backing the decode,
+// so Close must be called to release them.
+type Stream struct {
+ // compressed is retained so the caller's buffer cannot be collected
+ // while the memory stream built from it is still alive.
+ compressed []byte
+
+ istream *IStream
+ byteStream *IMFByteStream
+ attributes *IMFAttributes
+ reader *IMFSourceReader
+ samples *SourceReader
+
+ format Format
+ closed bool
+}
+
+// Open prepares a decode of compressed audio held in memory. The
+// returned Stream yields s16le PCM and must be closed.
+func Open(compressed []byte) (_ *Stream, err error) {
+ s := &Stream{compressed: compressed}
+
+ // Anything already allocated is released if a later step fails.
+ defer func() {
+ if err != nil {
+ s.release()
+ }
+ }()
+
+ s.istream = SHCreateMemStream(unsafe.SliceData(compressed), len(compressed))
+ if s.istream == nil {
+ return nil, fmt.Errorf("could not allocate IStream")
}
- defer stream.Release()
// We need to adapt the generic IStream to a Media Foundation stream type.
- var mfByteStream *IMFByteStream
- if err := MFCreateMFByteStreamOnStream(stream, &mfByteStream); err != nil {
- return nil, format, fmt.Errorf("creating MFByteStream from IStream: %w", err)
+ if err := MFCreateMFByteStreamOnStream(s.istream, &s.byteStream); err != nil {
+ return nil, fmt.Errorf("creating MFByteStream from IStream: %w", err)
}
- defer mfByteStream.Release()
// Attributes to configure the source reader with; specifically, enable hardware codecs.
- var attributes *IMFAttributes
- if err := MFCreateAttributes(&attributes, 1); err != nil {
- return nil, format, fmt.Errorf("creating attributes to apply to source reader: %w", err)
+ if err := MFCreateAttributes(&s.attributes, 1); err != nil {
+ return nil, fmt.Errorf("creating attributes to apply to source reader: %w", err)
}
- defer attributes.Release()
-
- if err := attributes.SetUINT32(&MF_READWRITE_ENABLE_HARDWARE_TRANSFORMS, 1); err != nil {
- return nil, format, fmt.Errorf("enabling hardware transforms: %w", err)
+ if err := s.attributes.SetUINT32(&MF_READWRITE_ENABLE_HARDWARE_TRANSFORMS, 1); err != nil {
+ return nil, fmt.Errorf("enabling hardware transforms: %w", err)
}
// Create the source reader using the byte stream.
- var mfSourceReader *IMFSourceReader
- if err := MFCreateSourceReaderFromByteStream(mfByteStream, attributes, &mfSourceReader); err != nil {
- return nil, format, fmt.Errorf("creating IMFSourceReader from IMFByteStream: %w", err)
+ if err := MFCreateSourceReaderFromByteStream(s.byteStream, s.attributes, &s.reader); err != nil {
+ return nil, fmt.Errorf("creating IMFSourceReader from IMFByteStream: %w", err)
}
- defer mfSourceReader.Release()
- if err := configureAudioStream(mfSourceReader); err != nil {
- return nil, format, fmt.Errorf("configuring audio stream: %w", err)
+ if err := configureAudioStream(s.reader); err != nil {
+ return nil, fmt.Errorf("configuring audio stream: %w", err)
}
- format, err = getSourceReaderFormat(mfSourceReader)
+ s.format, err = getSourceReaderFormat(s.reader)
if err != nil {
- return nil, format, fmt.Errorf("getting format: %w", err)
+ return nil, fmt.Errorf("getting format: %w", err)
+ }
+
+ s.samples = NewSourceReader(s.reader)
+
+ return s, nil
+}
+
+// Format describes the PCM this stream produces. It is known as soon as
+// the stream is opened.
+func (s *Stream) Format() Format {
+ return s.format
+}
+
+// Read fills p with decoded PCM, returning [io.EOF] once the source is
+// exhausted.
+func (s *Stream) Read(p []byte) (int, error) {
+ if s.closed {
+ return 0, fmt.Errorf("read on closed stream")
+ }
+ n, err := s.samples.Read(p)
+ runtime.KeepAlive(s.compressed)
+ return n, err
+}
+
+// Close releases the Media Foundation objects backing the stream. It is
+// idempotent.
+func (s *Stream) Close() error {
+ if s.closed {
+ return nil
+ }
+ s.closed = true
+ s.release()
+ return nil
+}
+
+// release drops every object the stream holds, in reverse order of
+// acquisition. Safe to call on a partially constructed Stream.
+func (s *Stream) release() {
+ if s.samples != nil {
+ s.samples.Close()
+ s.samples = nil
+ }
+ if s.reader != nil {
+ s.reader.Release()
+ s.reader = nil
+ }
+ if s.attributes != nil {
+ s.attributes.Release()
+ s.attributes = nil
+ }
+ if s.byteStream != nil {
+ s.byteStream.Release()
+ s.byteStream = nil
}
+ if s.istream != nil {
+ s.istream.Release()
+ s.istream = nil
+ }
+ runtime.KeepAlive(s.compressed)
+ s.compressed = nil
+}
- buf, err := io.ReadAll(NewSourceReader(mfSourceReader))
+// Decode compressed audio, returning the uncompressed data as PCM data
+// (s16le) and details about the PCM required to playback correctly.
+func Decode(compressed []byte) (uncompressed []byte, format Format, err error) {
+ s, err := Open(compressed)
if err != nil {
return nil, format, err
}
+ defer s.Close()
- runtime.KeepAlive(compressed)
+ buf, err := io.ReadAll(s)
+ if err != nil {
+ return nil, s.format, err
+ }
- return buf, format, nil
+ return buf, s.format, nil
}
// configureAudioStream selects the first audio stream and configures it output PCM.
@@ -291,6 +377,24 @@ func (s *SourceReader) copyInto(p []byte) int {
return n
}
+// Close releases any sample the reader is still holding. A stream that
+// is abandoned before EOF leaves a locked buffer and a live sample
+// behind, so this is not merely tidiness.
+func (s *SourceReader) Close() error {
+ if s.buffer != nil {
+ s.buffer.Unlock()
+ s.buffer.Release()
+ s.buffer = nil
+ }
+ if s.sample != nil {
+ s.sample.Release()
+ s.sample = nil
+ }
+ s.data = nil
+ s.chunk = nil
+ return nil
+}
+
/*
The following contains the minimal set of definitions we need to decode
audio using Media Framework.
diff --git a/internal/test/stream_test.go b/internal/test/stream_test.go
@@ -0,0 +1,226 @@
+package test
+
+import (
+ "bytes"
+ "io"
+ "testing"
+ "time"
+
+ "git.sr.ht/~jackmordaunt/nativeaudio"
+)
+
+// TestStreamMatchesDecode ensures the incremental path produces exactly
+// the same PCM as the buffered one. Any divergence here means the two
+// code paths have drifted apart.
+func TestStreamMatchesDecode(t *testing.T) {
+ d := newDecoder(t)
+
+ want, wantFormat, err := d.Decode(compressed)
+ if err != nil {
+ t.Fatalf("Decode: %v", err)
+ }
+
+ s, err := d.Stream(compressed)
+ if err != nil {
+ t.Fatalf("Stream: %v", err)
+ }
+ defer s.Close()
+
+ // The format must be known before a single byte is read.
+ if s.Format() != wantFormat {
+ t.Errorf("stream format: want %+v, got %+v", wantFormat, s.Format())
+ }
+
+ got, err := io.ReadAll(s)
+ if err != nil {
+ t.Fatalf("reading stream: %v", err)
+ }
+ if !bytes.Equal(got, want) {
+ t.Fatalf("stream output differs from buffered decode: %d vs %d bytes", len(got), len(want))
+ }
+}
+
+// TestStreamFileMatchesDecodeFile covers the file entry point, which on
+// some backends pipes the decode rather than buffering it.
+func TestStreamFileMatchesDecodeFile(t *testing.T) {
+ d := newDecoder(t)
+
+ want, wantFormat, err := d.DecodeFile("compressed.m4a")
+ if err != nil {
+ t.Fatalf("DecodeFile: %v", err)
+ }
+
+ s, err := d.StreamFile("compressed.m4a")
+ if err != nil {
+ t.Fatalf("StreamFile: %v", err)
+ }
+ defer s.Close()
+
+ if s.Format() != wantFormat {
+ t.Errorf("stream format: want %+v, got %+v", wantFormat, s.Format())
+ }
+
+ got, err := io.ReadAll(s)
+ if err != nil {
+ t.Fatalf("reading stream: %v", err)
+ }
+ if !bytes.Equal(got, want) {
+ t.Fatalf("stream output differs from buffered decode: %d vs %d bytes", len(got), len(want))
+ }
+}
+
+// TestStreamSmallReads ensures the stream honours whatever buffer size
+// the caller offers, rather than assuming it is handed a large one.
+func TestStreamSmallReads(t *testing.T) {
+ d := newDecoder(t)
+
+ want, _, err := d.Decode(compressed)
+ if err != nil {
+ t.Fatalf("Decode: %v", err)
+ }
+
+ s, err := d.Stream(compressed)
+ if err != nil {
+ t.Fatalf("Stream: %v", err)
+ }
+ defer s.Close()
+
+ var (
+ got [][]byte
+ buf = make([]byte, 7) // deliberately small and not sample-aligned
+ )
+ for {
+ n, err := s.Read(buf)
+ if n > 0 {
+ got = append(got, append([]byte(nil), buf[:n]...))
+ }
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ t.Fatalf("reading stream: %v", err)
+ }
+ }
+ if joined := bytes.Join(got, nil); !bytes.Equal(joined, want) {
+ t.Fatalf("small reads produced %d bytes, want %d", len(joined), len(want))
+ }
+}
+
+// TestStreamAbandonedEarly closes a stream without draining it. Backends
+// hold decoder state and, where ffmpeg is used, a live subprocess, so
+// this must not leak either.
+func TestStreamAbandonedEarly(t *testing.T) {
+ d := newDecoder(t)
+ for i := 0; i < 5; i++ {
+ s, err := d.Stream(compressed)
+ if err != nil {
+ t.Fatalf("Stream: %v", err)
+ }
+ if _, err := io.CopyN(io.Discard, s, 1024); err != nil {
+ t.Fatalf("partial read: %v", err)
+ }
+ if err := s.Close(); err != nil {
+ t.Fatalf("closing abandoned stream: %v", err)
+ }
+ if err := s.Close(); err != nil {
+ t.Fatalf("second close should be a no-op: %v", err)
+ }
+ }
+}
+
+// TestCloseWaitsForStreams ensures the decoder does not tear down
+// platform state while a stream is still using it.
+func TestCloseWaitsForStreams(t *testing.T) {
+ d, err := nativeaudio.New()
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ s, err := d.Stream(compressed)
+ if err != nil {
+ t.Fatalf("Stream: %v", err)
+ }
+
+ closed := make(chan error, 1)
+ go func() { closed <- d.Close() }()
+
+ select {
+ case err := <-closed:
+ t.Fatalf("Close returned while a stream was open: %v", err)
+ case <-time.After(100 * time.Millisecond):
+ }
+
+ if err := s.Close(); err != nil {
+ t.Fatalf("closing stream: %v", err)
+ }
+
+ select {
+ case err := <-closed:
+ if err != nil {
+ t.Fatalf("Close: %v", err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("Close did not return after the stream was closed")
+ }
+
+ if _, err := d.Stream(compressed); err != nativeaudio.ErrClosed {
+ t.Fatalf("stream after close: want ErrClosed, got %v", err)
+ }
+}
+
+// TestFFmpegStream covers the piped ffmpeg decode directly, on every
+// platform that has ffmpeg rather than only where it is the backend.
+func TestFFmpegStream(t *testing.T) {
+ requireTools(t, "ffmpeg", "ffprobe")
+
+ want, wantFormat, err := nativeaudio.FFmpegLoad("compressed.m4a")
+ if err != nil {
+ t.Fatalf("FFmpegLoad: %v", err)
+ }
+
+ s, err := nativeaudio.FFmpegStream("compressed.m4a")
+ if err != nil {
+ t.Fatalf("FFmpegStream: %v", err)
+ }
+ defer s.Close()
+
+ if s.Format() != wantFormat {
+ t.Errorf("stream format: want %+v, got %+v", wantFormat, s.Format())
+ }
+
+ got, err := io.ReadAll(s)
+ if err != nil {
+ t.Fatalf("reading stream: %v", err)
+ }
+ if !bytes.Equal(got, want) {
+ t.Fatalf("piped ffmpeg output differs from buffered: %d vs %d bytes", len(got), len(want))
+ }
+}
+
+// TestFFmpegStreamAbandoned kills the ffmpeg process early. There is no
+// portable way to inspect the process table, so the timeout is what
+// catches a stream that fails to reap its child.
+func TestFFmpegStreamAbandoned(t *testing.T) {
+ requireTools(t, "ffmpeg", "ffprobe")
+ for i := 0; i < 3; i++ {
+ s, err := nativeaudio.FFmpegStream("compressed.m4a")
+ if err != nil {
+ t.Fatalf("FFmpegStream: %v", err)
+ }
+ if _, err := io.CopyN(io.Discard, s, 512); err != nil {
+ t.Fatalf("partial read: %v", err)
+ }
+ done := make(chan error, 1)
+ go func() { done <- s.Close() }()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("closing abandoned ffmpeg stream: %v", err)
+ }
+ case <-time.After(30 * time.Second):
+ // Generous on purpose. This guards against a stream that never
+ // reaps its child at all, not against a slow one: a loaded CI
+ // runner took ten seconds just to tear the process down.
+ t.Fatal("closing an abandoned ffmpeg stream hung")
+ }
+ }
+}
diff --git a/stream.go b/stream.go
@@ -0,0 +1,58 @@
+package nativeaudio
+
+import (
+ "bytes"
+ "io"
+)
+
+// Stream decodes audio incrementally, yielding signed 16-bit
+// little-endian PCM through [io.Reader].
+//
+// It exists because the buffered API holds the whole decode in memory,
+// and PCM is roughly an order of magnitude larger than the compressed
+// input it came from. A few minutes of stereo audio is tens of
+// megabytes, which is fine for a sound effect and wasteful for a track.
+//
+// A Stream must be closed. Closing the Decoder that produced it waits
+// for every open Stream to be closed first.
+type Stream struct {
+ r io.ReadCloser
+ format Format
+ done func()
+}
+
+// Format describes the PCM this Stream produces. It is known as soon as
+// the Stream is opened, before any audio is read.
+func (s *Stream) Format() Format {
+ return s.format
+}
+
+// Read fills p with decoded PCM, returning [io.EOF] when the audio is
+// exhausted.
+func (s *Stream) Read(p []byte) (int, error) {
+ return s.r.Read(p)
+}
+
+// Close releases the decoder state behind the Stream. It is safe to
+// call more than once, and abandoning a Stream before [io.EOF] is fine
+// as long as it is closed.
+func (s *Stream) Close() error {
+ err := s.r.Close()
+ if s.done != nil {
+ s.done()
+ s.done = nil
+ }
+ return err
+}
+
+// newBufferedStream serves PCM that has already been fully decoded.
+//
+// Backends that cannot yet decode incrementally use this so that the
+// streaming API behaves correctly everywhere, at the cost of the memory
+// saving on those platforms.
+func newBufferedStream(pcm []byte, format Format) *Stream {
+ return &Stream{
+ r: io.NopCloser(bytes.NewReader(pcm)),
+ format: format,
+ }
+}
