playback: apply the gain at the media item flush

The app-thread transition callback was only ever a stopgap: media3 posts
it after the playback thread has already rendered, so the first buffers
of a new item could be scaled by the previous item's gain.

media3 forwards the playback timeline and the current period to
AudioProcessor.flush, and DefaultAudioSink flushes the processors on
every sink configuration, which happens at each media item transition
even when the format is unchanged. That is early enough to set the gain
before any of the new item's buffers are processed, on the playback
thread. See androidx/media#418.

The processor resolves the period back to its MediaItem and reports it,
so the gain maths and the preferences stay in ReplayGainController. That
controller is now written from two threads, so its state is volatile.

The transition callback stays as the fallback for a flush that carries no
period: without it a stale gain would persist for a whole track rather
than for a few milliseconds.

app/src/main/kotlin/com/subtracks/playback/PlaybackService.kt

@@ -62,8 +62,8 @@ class PlaybackService : MediaSessionService() {
     private var starred = false
     private var songJob: Job? = null
 
-    private val gainProcessor = ReplayGainProcessor()
-    private val replayGain = ReplayGainController { gainProcessor.setLinearGain(it) }
+    private val gainProcessor: ReplayGainProcessor = ReplayGainProcessor { replayGain.applyItem(it) }
+    private val replayGain: ReplayGainController = ReplayGainController { gainProcessor.setLinearGain(it) }
 
     private val starCallback =
         object : MediaSession.Callback {

app/src/main/kotlin/com/subtracks/playback/ReplayGainController.kt

@@ -5,7 +5,12 @@ import com.subtracks.data.prefs.ReplayGainMode
 internal class ReplayGainController(
     private val apply: (Float) -> Unit,
 ) {
+    // Written from the playback thread by ReplayGainProcessor.onFlush and from the app thread by
+    // the media item transition fallback.
+    @Volatile
     private var track = ReplayGain()
+
+    @Volatile
     private var settings = ReplayGainSettings(ReplayGainMode.Track, 0f)
 
     fun setTrack(value: ReplayGain?) {

app/src/main/kotlin/com/subtracks/playback/ReplayGainProcessor.kt

@@ -2,14 +2,19 @@ package com.subtracks.playback
 
 import androidx.annotation.OptIn
 import androidx.media3.common.C
+import androidx.media3.common.MediaItem
+import androidx.media3.common.Timeline
 import androidx.media3.common.audio.AudioProcessor.AudioFormat
+import androidx.media3.common.audio.AudioProcessor.StreamMetadata
 import androidx.media3.common.audio.BaseAudioProcessor
 import androidx.media3.common.util.UnstableApi
 import java.nio.ByteBuffer
 import kotlin.math.roundToInt
 
 @OptIn(UnstableApi::class)
-internal class ReplayGainProcessor : BaseAudioProcessor() {
+internal class ReplayGainProcessor(
+    private val onItem: (MediaItem?) -> Unit = {},
+) : BaseAudioProcessor() {
     @Volatile
     private var linearGain = 1f
 
@@ -19,6 +24,15 @@ internal class ReplayGainProcessor : BaseAudioProcessor() {
 
     override fun onConfigure(inputAudioFormat: AudioFormat): AudioFormat = inputAudioFormat
 
+    // DefaultAudioSink configures and flushes the processors at every media item transition, even
+    // when the format is unchanged, and hands over the timeline and the period being flushed. That
+    // is early enough to set the new item's gain before any of its buffers are processed, unlike
+    // Player.Listener.onMediaItemTransition, which the app thread receives afterwards.
+    // See androidx/media#418.
+    override fun onFlush(streamMetadata: StreamMetadata) {
+        onItem(streamMetadata.currentItem())
+    }
+
     override fun queueInput(inputBuffer: ByteBuffer) {
         val output = replaceOutputBuffer(inputBuffer.remaining())
         val gain = linearGain.takeIf { it.isFinite() } ?: 1f
@@ -46,3 +60,12 @@ internal class ReplayGainProcessor : BaseAudioProcessor() {
         const val MAX_SAMPLE = 32767
     }
 }
+
+@OptIn(UnstableApi::class)
+internal fun StreamMetadata.currentItem(): MediaItem? {
+    val uid = periodUid ?: return null
+    val index = timeline.getIndexOfPeriod(uid)
+    if (index == C.INDEX_UNSET) return null
+    val period = timeline.getPeriod(index, Timeline.Period())
+    return timeline.getWindow(period.windowIndex, Timeline.Window()).mediaItem
+}

app/src/test/kotlin/com/subtracks/playback/ReplayGainProcessorTest.kt

@@ -2,15 +2,21 @@ package com.subtracks.playback
 
 import androidx.annotation.OptIn
 import androidx.media3.common.C
+import androidx.media3.common.MediaItem
 import androidx.media3.common.audio.AudioProcessor.AudioFormat
 import androidx.media3.common.audio.AudioProcessor.StreamMetadata
 import androidx.media3.common.util.UnstableApi
+import androidx.media3.exoplayer.source.SinglePeriodTimeline
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.subtracks.data.prefs.ReplayGainMode
 import org.junit.Assert.assertEquals
 import org.junit.Test
+import org.junit.runner.RunWith
 import java.nio.ByteBuffer
 import java.nio.ByteOrder
 
 @OptIn(UnstableApi::class)
+@RunWith(AndroidJUnit4::class)
 class ReplayGainProcessorTest {
     @Test
     fun scalesSixteenBitSamples() {
@@ -85,6 +91,42 @@ class ReplayGainProcessorTest {
         assertEquals(true, processor.isActive())
     }
 
+    @Test
+    fun aFlushReportsTheItemBeingFlushed() {
+        val reported = mutableListOf<MediaItem?>()
+        val item = item("a", trackGain = -6.0206f)
+
+        ReplayGainProcessor { reported += it }.flush(streamMetadata(item))
+
+        assertEquals(listOf(item), reported)
+    }
+
+    @Test
+    fun aFlushWithoutAPeriodReportsNoItem() {
+        val reported = mutableListOf<MediaItem?>()
+
+        ReplayGainProcessor { reported += it }.flush(StreamMetadata.DEFAULT)
+
+        assertEquals(listOf(null), reported)
+    }
+
+    @Test
+    fun aFlushMovesTheGainToTheNextItem() {
+        lateinit var processor: ReplayGainProcessor
+        val controller = ReplayGainController { processor.setLinearGain(it) }
+        processor = ReplayGainProcessor { controller.applyItem(it) }
+        controller.setSettings(ReplayGainSettings(ReplayGainMode.Track, 0f))
+        processor.configure(AudioFormat(SAMPLE_RATE, CHANNELS, C.ENCODING_PCM_16BIT))
+
+        processor.flush(streamMetadata(item("a", trackGain = -6.0206f)))
+        processor.queueInput(sixteenBit(1000))
+        assertEquals(listOf(500), readShorts(processor.getOutput()))
+
+        processor.flush(streamMetadata(item("b", trackGain = 6.0206f)))
+        processor.queueInput(sixteenBit(1000))
+        assertEquals(listOf(2000), readShorts(processor.getOutput()))
+    }
+
     private fun processor(
         gain: Float,
         encoding: Int = C.ENCODING_PCM_16BIT,
@@ -95,6 +137,33 @@ class ReplayGainProcessorTest {
             flush(StreamMetadata.DEFAULT)
         }
 
+    private fun item(
+        id: String,
+        trackGain: Float,
+    ): MediaItem =
+        queueMediaItem(
+            QueueItem(
+                id = id,
+                title = id,
+                artist = null,
+                album = null,
+                coverArtId = null,
+                replayGain = ReplayGain(trackGain = trackGain),
+            ),
+            local = null,
+            stream = "http://server/$id",
+            transcode = false,
+        )
+
+    private fun streamMetadata(item: MediaItem): StreamMetadata {
+        val timeline = SinglePeriodTimeline(DURATION_US, false, false, false, null, item)
+        return StreamMetadata
+            .Builder()
+            .setTimeline(timeline)
+            .setPeriodUid(timeline.getUidOfPeriod(0))
+            .build()
+    }
+
     private fun sixteenBit(vararg samples: Int): ByteBuffer =
         ByteBuffer.allocate(samples.size * 2).order(ByteOrder.nativeOrder()).apply {
             samples.forEach { putShort(it.toShort()) }
@@ -125,5 +194,6 @@ class ReplayGainProcessorTest {
     private companion object {
         const val SAMPLE_RATE = 44_100
         const val CHANNELS = 2
+        const val DURATION_US = 1_000_000L
     }
 }