fm-rds-tx/internal/ingest/decoder/mp3/decoder.go

package mp3

import (
	"context"
	"encoding/binary"
	"fmt"
	"io"
	"time"

	gomp3 "github.com/hajimehoshi/go-mp3"
	"github.com/jan/fm-rds-tx/internal/ingest"
	"github.com/jan/fm-rds-tx/internal/ingest/decoder"
)

type Decoder struct{}

func New() *Decoder { return &Decoder{} }

func (d *Decoder) Name() string { return "mp3-native" }

func (d *Decoder) DecodeStream(ctx context.Context, r io.Reader, meta decoder.StreamMeta, emit func(ingest.PCMChunk) error) error {
	if r == nil {
		return fmt.Errorf("%w: mp3 decoder stream reader is nil", decoder.ErrUnsupported)
	}
	if emit == nil {
		return fmt.Errorf("%w: mp3 decoder emit callback is nil", decoder.ErrUnsupported)
	}

	dec, err := gomp3.NewDecoder(r)
	if err != nil {
		return fmt.Errorf("%w: mp3 decoder init: %v", decoder.ErrUnsupported, err)
	}

	const channels = 2 // go-mp3 always decodes to stereo s16le
	sampleRate := dec.SampleRate()
	if sampleRate <= 0 {
		if meta.SampleRateHz > 0 {
			sampleRate = meta.SampleRateHz
		} else {
			sampleRate = 44100
		}
	}

	const chunkFrames = 1024
	const frameBytes = channels * 2
	buf := make([]byte, chunkFrames*frameBytes)
	seq := uint64(0)

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}

		n, readErr := io.ReadAtLeast(dec, buf, frameBytes)
		if readErr != nil {
			if readErr == io.EOF || readErr == io.ErrUnexpectedEOF {
				if n > 0 {
					if err := emitChunk(buf[:n], seq, sampleRate, meta.SourceID, emit); err != nil {
						return err
					}
				}
				return nil
			}
			return fmt.Errorf("mp3 decoder read pcm: %w", readErr)
		}

		if err := emitChunk(buf[:n], seq, sampleRate, meta.SourceID, emit); err != nil {
			return err
		}
		seq++
	}
}

func emitChunk(data []byte, seq uint64, sampleRate int, sourceID string, emit func(ingest.PCMChunk) error) error {
	samples := make([]int32, 0, len(data)/2)
	for i := 0; i+1 < len(data); i += 2 {
		v := int16(binary.LittleEndian.Uint16(data[i : i+2]))
		samples = append(samples, int32(v)<<16)
	}
	return emit(ingest.PCMChunk{
		Samples:      samples,
		Channels:     2,
		SampleRateHz: sampleRate,
		Sequence:     seq,
		Timestamp:    time.Now(),
		SourceID:     sourceID,
	})
}