feat(android): AudioPrefetcher pre-downloads next-N tracks into SimpleCache
android / Build + lint + test (push) Failing after 1m31s
android / Build + lint + test (push) Failing after 1m31s
CacheSettings.prefetchWindow has shipped at default=5 since M8 but the prefetcher itself was deferred to a follow-up. Without it, forward skips re-fetch from the network every time and gapless transitions stall. New AudioPrefetcher singleton subscribes to PlayerController.uiState + AuthStore.cacheSettings, walks queue[currentIndex+1 .. +window], and runs each missing track through Media3's CacheWriter against the same SimpleCache the player reads from (PlayerFactory.simpleCache). DataSpec uses setKey(trackId) to match PlayerController.toMediaItem's setCustomCacheKey(id) — without this the player would miss the cached bytes on read-through. Reconcile is idempotent: CacheWriter is a no-op when bytes are already resident, so distinctUntilChanged on (queue, index, window) gates re-runs and the per-item check is cheap. Window slides cancel in-flight jobs for tracks that have dropped out (skip-prev, queue rebuild) so a stale prefetch doesn't keep the network busy. Eager-constructed via the construct-the-singleton trick in MinstrelApplication, alongside CacheIndexer and CoverPrefetcher. Mirrors flutter_client/lib/cache/prefetcher.dart's user-visible behavior (queue-walk + per-track pin + idempotent reconcile) implemented with the native Media3 primitives (CacheWriter + SimpleCache) instead of Flutter's AudioCacheManager.pin.
This commit is contained in:
@@ -13,6 +13,7 @@ import com.fabledsword.minstrel.di.ApplicationScope
|
||||
import com.fabledsword.minstrel.events.EventsStream
|
||||
import com.fabledsword.minstrel.events.LiveEventsDispatcher
|
||||
import com.fabledsword.minstrel.metadata.FreshnessSweeper
|
||||
import com.fabledsword.minstrel.player.AudioPrefetcher
|
||||
import com.fabledsword.minstrel.player.CoverPrefetcher
|
||||
import com.fabledsword.minstrel.player.PlayEventsReporter
|
||||
import com.fabledsword.minstrel.player.PlaybackErrorReporter
|
||||
@@ -103,6 +104,15 @@ class MinstrelApplication :
|
||||
*/
|
||||
@Suppress("unused") @Inject lateinit var cacheIndexer: CacheIndexer
|
||||
|
||||
/**
|
||||
* Same construct-the-singleton trick — AudioPrefetcher's init
|
||||
* block subscribes to PlayerController.uiState +
|
||||
* AuthStore.cacheSettings and pins the next-N tracks into
|
||||
* SimpleCache via CacheWriter. Without this @Inject the
|
||||
* prefetchWindow setting would be inert and skips would re-fetch.
|
||||
*/
|
||||
@Suppress("unused") @Inject lateinit var audioPrefetcher: AudioPrefetcher
|
||||
|
||||
/**
|
||||
* Same construct-the-singleton trick — VersionCheckController's
|
||||
* init block starts a 5-min poll loop against /healthz so the
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
package com.fabledsword.minstrel.player
|
||||
|
||||
import android.net.Uri
|
||||
import androidx.media3.datasource.DataSpec
|
||||
import androidx.media3.datasource.cache.CacheDataSource
|
||||
import androidx.media3.datasource.cache.CacheWriter
|
||||
import androidx.media3.datasource.okhttp.OkHttpDataSource
|
||||
import com.fabledsword.minstrel.auth.AuthStore
|
||||
import com.fabledsword.minstrel.di.ApplicationScope
|
||||
import com.fabledsword.minstrel.shared.resolveServerUrl
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.flow.combine
|
||||
import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
import kotlinx.coroutines.flow.map
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import okhttp3.OkHttpClient
|
||||
import javax.inject.Inject
|
||||
import javax.inject.Singleton
|
||||
|
||||
/**
|
||||
* Pre-downloads the next-N tracks in the queue into the shared Media3
|
||||
* [androidx.media3.datasource.cache.SimpleCache] so a skip-forward or
|
||||
* natural advance plays from disk instead of waiting on a fresh HTTP
|
||||
* connection. Mirrors the Flutter `Prefetcher` (cache/prefetcher.dart):
|
||||
* watches the player's current track, walks forward by
|
||||
* [com.fabledsword.minstrel.cache.audiocache.CacheSettings.prefetchWindow]
|
||||
* tracks, and pins each one. Idempotent — `CacheWriter` is a no-op when
|
||||
* a track is already fully resident, so the reconcile is cheap on
|
||||
* every queue/index change.
|
||||
*
|
||||
* Eager-constructed via the construct-the-singleton trick in
|
||||
* [com.fabledsword.minstrel.MinstrelApplication]; the init block wires
|
||||
* the observer.
|
||||
*/
|
||||
@Singleton
|
||||
class AudioPrefetcher @Inject constructor(
|
||||
@ApplicationScope private val scope: CoroutineScope,
|
||||
private val playerController: PlayerController,
|
||||
private val playerFactory: PlayerFactory,
|
||||
private val authStore: AuthStore,
|
||||
private val okHttpClient: OkHttpClient,
|
||||
) {
|
||||
// Reuse the same cache + upstream chain the player uses. Without the
|
||||
// SimpleCache reference here the prefetcher would write to a
|
||||
// disjoint store and playback would never read what we pinned.
|
||||
private val cacheDataSourceFactory: CacheDataSource.Factory =
|
||||
CacheDataSource.Factory()
|
||||
.setCache(playerFactory.simpleCache)
|
||||
.setUpstreamDataSourceFactory(OkHttpDataSource.Factory(okHttpClient))
|
||||
|
||||
// trackId -> in-flight job. Reconcile cancels jobs whose track has
|
||||
// dropped out of the window (skip-prev, queue rebuild) so a stale
|
||||
// prefetch doesn't keep the network busy.
|
||||
private val activeJobs = mutableMapOf<String, Job>()
|
||||
private val mutex = Mutex()
|
||||
|
||||
init {
|
||||
scope.launch {
|
||||
combine(
|
||||
playerController.uiState.map { it.queue.map { t -> t.id to t.streamUrl } },
|
||||
playerController.uiState.map { it.queueIndex },
|
||||
authStore.cacheSettings.map { it.prefetchWindow },
|
||||
) { queue, index, window -> Triple(queue, index, window) }
|
||||
.distinctUntilChanged()
|
||||
.collect { (queue, index, window) -> reconcile(queue, index, window) }
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun reconcile(
|
||||
queue: List<Pair<String, String>>,
|
||||
index: Int,
|
||||
window: Int,
|
||||
) {
|
||||
mutex.withLock {
|
||||
if (index < 0 || queue.isEmpty() || window <= 0) {
|
||||
cancelAllLocked()
|
||||
return
|
||||
}
|
||||
// Exclude the currently-playing track (it's loaded by the
|
||||
// player itself) and walk `window` tracks forward.
|
||||
val firstIdx = (index + 1).coerceAtMost(queue.size)
|
||||
val lastIdx = (index + window).coerceAtMost(queue.size - 1)
|
||||
if (firstIdx > lastIdx) {
|
||||
cancelAllLocked()
|
||||
return
|
||||
}
|
||||
val targets = queue.subList(firstIdx, lastIdx + 1)
|
||||
val targetIds = targets.mapTo(mutableSetOf()) { it.first }
|
||||
|
||||
// Cancel jobs for tracks that have slid out of the window.
|
||||
activeJobs.entries
|
||||
.filter { it.key !in targetIds }
|
||||
.toList()
|
||||
.forEach { (id, job) ->
|
||||
job.cancel()
|
||||
activeJobs.remove(id)
|
||||
}
|
||||
|
||||
// Start prefetches for new arrivals. Skip blank URLs (these
|
||||
// come from minimal TrackRefs synthesized from playlist rows
|
||||
// when the upstream track was removed from the library).
|
||||
for ((trackId, streamUrl) in targets) {
|
||||
if (trackId in activeJobs) continue
|
||||
if (streamUrl.isBlank()) continue
|
||||
val job = scope.launch(Dispatchers.IO) {
|
||||
runCatching { prefetchOne(trackId, streamUrl) }
|
||||
mutex.withLock { activeJobs.remove(trackId) }
|
||||
}
|
||||
activeJobs[trackId] = job
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun cancelAllLocked() {
|
||||
activeJobs.values.forEach { it.cancel() }
|
||||
activeJobs.clear()
|
||||
}
|
||||
|
||||
private fun prefetchOne(trackId: String, streamUrl: String) {
|
||||
// Match PlayerController.toMediaItem's resolveServerUrl +
|
||||
// setCustomCacheKey(id) so the cached bytes are keyed exactly
|
||||
// like a normal playback request — otherwise the player would
|
||||
// miss them on read-through.
|
||||
val resolved = resolveServerUrl(streamUrl) ?: streamUrl
|
||||
val dataSpec = DataSpec.Builder()
|
||||
.setUri(Uri.parse(resolved))
|
||||
.setKey(trackId)
|
||||
.build()
|
||||
val writer = CacheWriter(
|
||||
cacheDataSourceFactory.createDataSource(),
|
||||
dataSpec,
|
||||
/* temporaryBuffer = */ null,
|
||||
/* progressListener = */ null,
|
||||
)
|
||||
// Blocking, but we're on Dispatchers.IO. Throws on network
|
||||
// failure — outer runCatching swallows so a single failed
|
||||
// prefetch doesn't trip the whole reconcile.
|
||||
writer.cache()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user