feat: integrate audio engine pipeline with MediaCodec and AudioTrack

Made-with: Cursor

app/src/main/java/xyz/cottongin/radio247/audio/AudioEngine.kt

@@ -0,0 +1,142 @@
+package xyz.cottongin.radio247.audio
+
+import android.media.AudioAttributes
+import android.media.AudioFormat
+import android.media.AudioTrack
+import android.media.MediaCodec
+import android.media.MediaFormat
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
+import java.util.concurrent.atomic.AtomicLong
+
+class AudioEngine(
+    private val url: String,
+    private val bufferMs: Int = 0
+) {
+    private val _events = MutableSharedFlow<AudioEngineEvent>(extraBufferCapacity = 64)
+    val events: SharedFlow<AudioEngineEvent> = _events
+
+    private var thread: Thread? = null
+    @Volatile
+    private var running = false
+    private val _estimatedLatencyMs = AtomicLong(0)
+
+    val estimatedLatencyMs: Long get() = _estimatedLatencyMs.get()
+
+    fun start() {
+        running = true
+        thread = Thread({
+            try {
+                runPipeline()
+            } catch (e: Exception) {
+                if (running) {
+                    val error = when (e) {
+                        is ConnectionFailed ->
+                            EngineError.ConnectionFailed(e)
+                        else -> EngineError.DecoderError(e)
+                    }
+                    _events.tryEmit(AudioEngineEvent.Error(error))
+                }
+            } finally {
+                _events.tryEmit(AudioEngineEvent.Stopped)
+            }
+        }, "AudioEngine").apply { start() }
+    }
+
+    fun stop() {
+        running = false
+        thread?.interrupt()
+        thread = null
+    }
+
+    private fun runPipeline() {
+        val connection = StreamConnection(url)
+        connection.open()
+
+        val sampleRate = 44100
+        val channelConfig = AudioFormat.CHANNEL_OUT_STEREO
+        val encoding = AudioFormat.ENCODING_PCM_16BIT
+        val minBuf = AudioTrack.getMinBufferSize(sampleRate, channelConfig, encoding)
+
+        val audioTrack = AudioTrack.Builder()
+            .setAudioAttributes(
+                AudioAttributes.Builder()
+                    .setUsage(AudioAttributes.USAGE_MEDIA)
+                    .setContentType(AudioAttributes.CONTENT_TYPE_MUSIC)
+                    .build()
+            )
+            .setAudioFormat(
+                AudioFormat.Builder()
+                    .setSampleRate(sampleRate)
+                    .setChannelMask(channelConfig)
+                    .setEncoding(encoding)
+                    .build()
+            )
+            .setBufferSizeInBytes(minBuf)
+            .setTransferMode(AudioTrack.MODE_STREAM)
+            .build()
+
+        val codec = MediaCodec.createDecoderByType("audio/mpeg")
+        val format = MediaFormat.createAudioFormat("audio/mpeg", sampleRate, 2)
+        codec.configure(format, null, null, 0)
+        codec.start()
+        audioTrack.play()
+
+        _events.tryEmit(AudioEngineEvent.Started)
+
+        try {
+            val bufferFrames = if (bufferMs > 0) (bufferMs / 26).coerceAtLeast(1) else 0
+
+            val ringBuffer = RingBuffer(bufferFrames) { mp3Frame ->
+                decodeToPcm(codec, mp3Frame, audioTrack)
+            }
+
+            val frameSync = Mp3FrameSync { mp3Frame ->
+                ringBuffer.write(mp3Frame)
+            }
+
+            val icyParser = IcyParser(
+                input = connection.inputStream!!,
+                metaint = connection.metaint,
+                onAudioData = { buf, off, len -> frameSync.feed(buf, off, len) },
+                onMetadata = { _events.tryEmit(AudioEngineEvent.MetadataChanged(it)) }
+            )
+
+            icyParser.readAll()
+
+            // Stream ended normally
+            ringBuffer.flush()
+            frameSync.flush()
+            _events.tryEmit(AudioEngineEvent.Error(EngineError.StreamEnded))
+        } finally {
+            codec.stop()
+            codec.release()
+            audioTrack.stop()
+            audioTrack.release()
+            connection.close()
+        }
+    }
+
+    private fun decodeToPcm(codec: MediaCodec, mp3Frame: ByteArray, audioTrack: AudioTrack) {
+        val inIdx = codec.dequeueInputBuffer(1000)
+        if (inIdx >= 0) {
+            val inBuf = codec.getInputBuffer(inIdx)!!
+            inBuf.clear()
+            inBuf.put(mp3Frame)
+            codec.queueInputBuffer(inIdx, 0, mp3Frame.size, 0, 0)
+        }
+
+        val bufferInfo = MediaCodec.BufferInfo()
+        var outIdx = codec.dequeueOutputBuffer(bufferInfo, 1000)
+        while (outIdx >= 0) {
+            val outBuf = codec.getOutputBuffer(outIdx)!!
+            outBuf.position(bufferInfo.offset)
+            outBuf.limit(bufferInfo.offset + bufferInfo.size)
+            val pcmData = ByteArray(bufferInfo.size)
+            outBuf.get(pcmData)
+            codec.releaseOutputBuffer(outIdx, false)
+            audioTrack.write(pcmData, 0, pcmData.size)
+            outIdx = codec.dequeueOutputBuffer(bufferInfo, 0)
+        }
+    }
+}

app/src/main/java/xyz/cottongin/radio247/audio/AudioEngineEvent.kt

@@ -0,0 +1,15 @@
+package xyz.cottongin.radio247.audio
+
+sealed interface AudioEngineEvent {
+    data class MetadataChanged(val metadata: IcyMetadata) : AudioEngineEvent
+    data class Error(val cause: EngineError) : AudioEngineEvent
+    data object Started : AudioEngineEvent
+    data object Stopped : AudioEngineEvent
+}
+
+sealed interface EngineError {
+    data class ConnectionFailed(val cause: Throwable) : EngineError
+    data object StreamEnded : EngineError
+    data class DecoderError(val cause: Throwable) : EngineError
+    data class AudioOutputError(val cause: Throwable) : EngineError
+}

app/src/main/java/xyz/cottongin/radio247/audio/RingBuffer.kt

@@ -0,0 +1,25 @@
+package xyz.cottongin.radio247.audio
+
+class RingBuffer(
+    private val capacityFrames: Int,
+    private val onFrame: (ByteArray) -> Unit
+) {
+    private val buffer = ArrayDeque<ByteArray>()
+
+    fun write(frame: ByteArray) {
+        if (capacityFrames == 0) {
+            onFrame(frame)
+            return
+        }
+        if (buffer.size >= capacityFrames) {
+            onFrame(buffer.removeFirst())
+        }
+        buffer.addLast(frame)
+    }
+
+    fun flush() {
+        while (buffer.isNotEmpty()) {
+            onFrame(buffer.removeFirst())
+        }
+    }
+}

app/src/test/java/xyz/cottongin/radio247/audio/RingBufferTest.kt

@@ -0,0 +1,61 @@
+package xyz.cottongin.radio247.audio
+
+import org.junit.Assert.assertArrayEquals
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+class RingBufferTest {
+
+    @Test
+    fun passthroughWhenCapacityIsZero() {
+        val emitted = mutableListOf<ByteArray>()
+        val buffer = RingBuffer(0) { emitted.add(it) }
+
+        buffer.write(byteArrayOf(1))
+        buffer.write(byteArrayOf(2))
+        buffer.write(byteArrayOf(3))
+
+        assertEquals(3, emitted.size)
+        assertArrayEquals(byteArrayOf(1), emitted[0])
+        assertArrayEquals(byteArrayOf(2), emitted[1])
+        assertArrayEquals(byteArrayOf(3), emitted[2])
+    }
+
+    @Test
+    fun buffersUpToCapacityThenEmitsOldest() {
+        val emitted = mutableListOf<ByteArray>()
+        val buffer = RingBuffer(2) { emitted.add(it) }
+
+        buffer.write(byteArrayOf(1))
+        assertEquals(0, emitted.size)
+
+        buffer.write(byteArrayOf(2))
+        assertEquals(0, emitted.size)
+
+        buffer.write(byteArrayOf(3))
+        assertEquals(1, emitted.size)
+        assertArrayEquals(byteArrayOf(1), emitted[0])
+    }
+
+    @Test
+    fun flushEmitsRemainingBufferedFrames() {
+        val emitted = mutableListOf<ByteArray>()
+        val buffer = RingBuffer(2) { emitted.add(it) }
+
+        buffer.write(byteArrayOf(1))
+        assertEquals(0, emitted.size)
+
+        buffer.flush()
+        assertEquals(1, emitted.size)
+        assertArrayEquals(byteArrayOf(1), emitted[0])
+    }
+
+    @Test
+    fun flushOnEmptyBufferDoesNothing() {
+        val emitted = mutableListOf<ByteArray>()
+        val buffer = RingBuffer(2) { emitted.add(it) }
+
+        buffer.flush()
+        assertEquals(0, emitted.size)
+    }
+}