mirror of
https://github.com/KRTirtho/spotube.git
synced 2026-11-04 10:13:59 +00:00
fix(jam): keep guests on the host's recording across song changes
- The host sends its source for the current song and the next two with every anchor, so guests switch to the same recording before the song starts. Guests drop the cached audio and stream URL for a re-pinned song and reload it. - Pins are retried once the room's queue arrives, instead of being lost when a guest joins before the queue. - A guest that finishes a song a little early waits for the host instead of jumping back to the old song, and guests don't start endless playback on their own. - Position tracking restarts on every song change, ignoring stale reports from the previous song. - The device queue keeps its queue and current-entry flows live all the time. With WhileSubscribed they went stale when nothing on screen watched them, and the stream proxy answered 404 for queued tracks. - Simulation tests run on the real device queue, cover recording switches, early song ends, idle CPU use and a long soak session.
This commit is contained in:
Kingkor Roy Tirtho
parent
aaa4d66db1
commit
ac5971846b
@ -44,6 +44,12 @@ interface AudioPlayerQueue {
|
||||
suspend fun getCurrentCollectionEntry(): QueueCollectionEntry?
|
||||
suspend fun getCollectionHistory(): List<QueueCollectionEntry>
|
||||
|
||||
/**
|
||||
* Stops endless playback from adding recommendations when the queue runs out. A jam
|
||||
* guest's queue belongs to the host; recommendations of its own would split them.
|
||||
*/
|
||||
fun setEndlessPlaybackSuppressed(suppressed: Boolean) = Unit
|
||||
|
||||
fun isPlaylistPlaying(playlistId: String? = null): Boolean {
|
||||
val entry = currentCollectionEntryFlow.value as? QueueCollectionEntry.Playlist ?: return false
|
||||
return playlistId == null || entry.id == playlistId
|
||||
|
||||
@ -57,19 +57,22 @@ class DeviceAudioPlayerQueue(
|
||||
|
||||
private var isRestoringPersistedState = true
|
||||
|
||||
// Eagerly: callers read `.value` (the stream proxy looks up the track it's asked to play,
|
||||
// Jam follows the room). With WhileSubscribed, `.value` went stale whenever nothing on
|
||||
// screen was collecting, so the proxy answered "not in the queue" (404) for new tracks.
|
||||
override val queueFlow: StateFlow<List<QueueEntry>> = combine(
|
||||
audioPlayer.playlistFlow,
|
||||
entryByMediaUrl) { playlist, knownEntries ->
|
||||
playlist.map { mediaItem ->
|
||||
knownEntries[mediaItem.url] ?: mediaItem.toFallbackQueueEntry()
|
||||
}
|
||||
}.stateIn(scope, SharingStarted.WhileSubscribed(5_000), emptyList())
|
||||
}.stateIn(scope, SharingStarted.Eagerly, emptyList())
|
||||
|
||||
override val currentQueueEntryFlow: StateFlow<QueueEntry?> = combine(
|
||||
audioPlayer.currentMediaItemFlow,
|
||||
entryByMediaUrl) { mediaItem, knownEntries ->
|
||||
mediaItem?.let { knownEntries[it.url] ?: it.toFallbackQueueEntry() }
|
||||
}.stateIn(scope, SharingStarted.WhileSubscribed(5_000), null)
|
||||
}.stateIn(scope, SharingStarted.Eagerly, null)
|
||||
|
||||
override val currentCollectionEntryFlow: StateFlow<QueueCollectionEntry?> =
|
||||
currentCollectionEntryState.asStateFlow()
|
||||
@ -306,9 +309,14 @@ class DeviceAudioPlayerQueue(
|
||||
}
|
||||
|
||||
private var isFetchingRecommendations = false
|
||||
private var endlessPlaybackSuppressed = false
|
||||
|
||||
override fun setEndlessPlaybackSuppressed(suppressed: Boolean) {
|
||||
endlessPlaybackSuppressed = suppressed
|
||||
}
|
||||
|
||||
private suspend fun handleQueueCompletion() {
|
||||
if (isFetchingRecommendations) return
|
||||
if (isFetchingRecommendations || endlessPlaybackSuppressed) return
|
||||
|
||||
val settings = settingsProvider.settingsState.value ?: return
|
||||
if (!settings.enableEndlessPlayback) return
|
||||
|
||||
@ -287,7 +287,11 @@ val sharedModules = module {
|
||||
}
|
||||
single { RemotePlaybackController(get(), get(), get(), get(), get()) }
|
||||
singleOf(::JamRoomClient)
|
||||
single { JamRoomService(get(), get(), get(), get(), get(), get(), get()) }
|
||||
single {
|
||||
JamRoomService(
|
||||
get(), get(), get(), get(), get(), get(), get(),
|
||||
evictCachedAudio = get<CacheManager>()::invalidateCacheEntry)
|
||||
}
|
||||
singleOf(::AudioPlayerQueueRepository) { bind<QueueStateRepository>() }
|
||||
single<AudioPlayerQueue> {
|
||||
DeviceAudioPlayerQueue(get(), get(), get(), get(), get())
|
||||
|
||||
@ -41,6 +41,10 @@ data class JamQueue(
|
||||
* time [serverTimeMs]. While [advancing], everyone extrapolates the position from it at
|
||||
* [rate]; a buffering host sends a non-advancing anchor so guests don't run ahead.
|
||||
*/
|
||||
/** The audio source the host plays [trackKey] from. */
|
||||
@Serializable
|
||||
data class JamTrackSource(val trackKey: String, val source: PluginTaggedAudioSource)
|
||||
|
||||
@Serializable
|
||||
data class JamAnchor(
|
||||
val trackKey: String,
|
||||
@ -52,7 +56,12 @@ data class JamAnchor(
|
||||
val advancing: Boolean = playing,
|
||||
val rate: Float = 1f,
|
||||
/** The audio source the host plays [trackKey] from, so guests stream the same file. */
|
||||
val source: PluginTaggedAudioSource? = null) {
|
||||
val source: PluginTaggedAudioSource? = null,
|
||||
/**
|
||||
* The host's sources for the next songs. A player loads the next song long before it
|
||||
* plays, so guests need the host's recording of it well before then.
|
||||
*/
|
||||
val upcoming: List<JamTrackSource> = emptyList()) {
|
||||
/** The room's position at server time [serverNowMs]. */
|
||||
fun positionAt(serverNowMs: Long): Long =
|
||||
if (advancing) positionMs + ((serverNowMs - serverTimeMs) * rate).toLong() else positionMs
|
||||
|
||||
@ -46,6 +46,8 @@ import kotlinx.coroutines.flow.SharingStarted
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.flow.combine
|
||||
import kotlinx.coroutines.flow.collectLatest
|
||||
import kotlinx.coroutines.flow.map
|
||||
import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
import kotlinx.coroutines.flow.launchIn
|
||||
import kotlinx.coroutines.flow.mapNotNull
|
||||
@ -94,7 +96,9 @@ class JamRoomService(
|
||||
private val settingsRepository: UserSettingsStore,
|
||||
private val audioPluginSource: AudioPluginSource,
|
||||
private val trackSources: TrackSourceRepository,
|
||||
private val streamingUrlRepository: StreamingUrlRepository) {
|
||||
private val streamingUrlRepository: StreamingUrlRepository,
|
||||
/** Deletes a track's audio from the disk cache (`CacheManager.invalidateCacheEntry`). */
|
||||
private val evictCachedAudio: suspend (trackId: String) -> Unit) {
|
||||
private val log = Logger.withTag("JamRoomService")
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
|
||||
@ -174,6 +178,11 @@ class JamRoomService(
|
||||
private var positionAdvancing = false
|
||||
private var ownSeekUntilMs = 0L
|
||||
private var entryChangedAtMs = 0L
|
||||
/** The song [positionSampleMs] belongs to; a new song starts the tracking from zero. */
|
||||
private var positionEntryKey: String? = null
|
||||
|
||||
/** Host: its sources for the next songs, sent with every anchor. */
|
||||
private var upcomingSources: List<JamTrackSource> = emptyList()
|
||||
|
||||
private val anchorRequests = Channel<Unit>(Channel.CONFLATED)
|
||||
|
||||
@ -193,7 +202,7 @@ class JamRoomService(
|
||||
.onEach { onEvent(it) }
|
||||
.launchIn(scope)
|
||||
audioPlayerQueue.currentQueueEntryFlow
|
||||
.onEach { entryChangedAtMs = now() }
|
||||
.onEach { onEntryChanged(it?.jamKey()) }
|
||||
.launchIn(scope)
|
||||
audioPlayer.positionFlow
|
||||
.onEach { onPosition(it.inWholeMilliseconds) }
|
||||
@ -294,6 +303,8 @@ class JamRoomService(
|
||||
|
||||
private suspend fun reset() {
|
||||
inRoom = false
|
||||
audioPlayerQueue.setEndlessPlaybackSuppressed(false)
|
||||
upcomingSources = emptyList()
|
||||
roleJob?.cancel()
|
||||
roleJob = null
|
||||
restoreSpeed()
|
||||
@ -407,7 +418,9 @@ class JamRoomService(
|
||||
|
||||
private suspend fun runHost() {
|
||||
restoreSpeed()
|
||||
audioPlayerQueue.setEndlessPlaybackSuppressed(false)
|
||||
sentAnchor = null
|
||||
upcomingSources = emptyList()
|
||||
kotlinx.coroutines.coroutineScope {
|
||||
combine(audioPlayerQueue.queueFlow, audioPlayer.shuffleModeFlow) { queue, shuffle ->
|
||||
JamQueue(queue.map(JamMediaItem::fromQueueEntry), shuffle)
|
||||
@ -443,6 +456,27 @@ class JamRoomService(
|
||||
.onEach { publishAnchor() }
|
||||
.launchIn(this)
|
||||
|
||||
// Share the next songs' sources as soon as a song starts: a guest's player loads
|
||||
// the next song long before it plays, and must find the host's recording by then.
|
||||
launch {
|
||||
combine(
|
||||
audioPlayerQueue.currentQueueEntryFlow.map { it?.jamKey() }.distinctUntilChanged(),
|
||||
audioPlayerQueue.queueFlow) { key, queue -> key to queue }
|
||||
.collectLatest { (key, queue) ->
|
||||
val start = queue.indexOfFirst { it.jamKey() == key } + 1
|
||||
if (start <= 0) return@collectLatest
|
||||
val upcoming = queue.drop(start).take(UPCOMING_SOURCES)
|
||||
.filterIsInstance<QueueEntry.StreamingTrack>()
|
||||
.mapNotNull { entry ->
|
||||
hostSource(entry.track, match = true)?.let { JamTrackSource(entry.track.id, it) }
|
||||
}
|
||||
if (upcoming != upcomingSources) {
|
||||
upcomingSources = upcoming
|
||||
publishAnchor()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Drift only builds up while playing; a paused host checks nothing.
|
||||
launch {
|
||||
while (isActive) {
|
||||
@ -491,6 +525,7 @@ class JamRoomService(
|
||||
private suspend fun sendAnchor() {
|
||||
if (_role.value != JamRole.Host) return
|
||||
val current = audioPlayerQueue.currentQueueEntryFlow.value ?: return
|
||||
onEntryChanged(current.jamKey())
|
||||
val playing = playIntent.value
|
||||
val next = JamAnchor(
|
||||
trackKey = current.jamKey(),
|
||||
@ -500,19 +535,29 @@ class JamRoomService(
|
||||
playing = playing,
|
||||
advancing = playing && audioPlayer.playerStateFlow.value == PlayerState.PLAYING,
|
||||
rate = audioPlayer.playbackSpeedFlow.value,
|
||||
source = currentSource(current))
|
||||
source = (current as? QueueEntry.StreamingTrack)?.let { hostSource(it.track, match = false) },
|
||||
upcoming = upcomingSources)
|
||||
sentAnchor = next
|
||||
sentAnchorAtMs = now()
|
||||
hostDrifts.clear()
|
||||
jamClient.sendAnchor(next)
|
||||
}
|
||||
|
||||
/** The source the host's audio plugin matched to [entry], so guests can use the same one. */
|
||||
private suspend fun currentSource(entry: QueueEntry): PluginTaggedAudioSource? {
|
||||
if (entry !is QueueEntry.StreamingTrack) return null
|
||||
/**
|
||||
* The source the host's audio plugin matched to [track], so guests can use the same one.
|
||||
* With [match], a track not matched yet is matched now (as the player would when it got
|
||||
* there); the current song is being matched by the player already.
|
||||
*/
|
||||
private suspend fun hostSource(track: MetadataTrack, match: Boolean): PluginTaggedAudioSource? {
|
||||
val plugin = audioPluginSource.selectedAudioPlugin.value ?: return null
|
||||
return runCatching { trackSources.getTrackSource(entry.track, plugin.pluginId) }.getOrNull()
|
||||
?.let { PluginTaggedAudioSource(plugin.pluginId, it) }
|
||||
suspend fun saved() = runCatching { trackSources.getTrackSource(track, plugin.pluginId) }.getOrNull()
|
||||
val source = saved() ?: if (match) {
|
||||
runCatching { streamingUrlRepository.resolveStreamInfo(track) }
|
||||
saved()
|
||||
} else {
|
||||
null
|
||||
}
|
||||
return source?.let { PluginTaggedAudioSource(plugin.pluginId, it) }
|
||||
}
|
||||
|
||||
private suspend fun applyCommand(command: JamCommand, from: String) {
|
||||
@ -550,7 +595,10 @@ class JamRoomService(
|
||||
// Guests follow the host's order through anchors; a local shuffle would only
|
||||
// make auto-advance pick a different track for a moment.
|
||||
runCatching { audioPlayer.shuffle(false) }
|
||||
audioPlayerQueue.setEndlessPlaybackSuppressed(true)
|
||||
corrector.reset()
|
||||
// The host's recordings first, so the queue's first load already uses them.
|
||||
anchor?.let { pinSources(it) }
|
||||
roomQueue?.let { applyQueue(it) }
|
||||
anchor?.let { onAnchor(it) }
|
||||
while (true) {
|
||||
@ -599,7 +647,12 @@ class JamRoomService(
|
||||
.onFailure { log.w(it) { "Failed to follow the host's play state" } }
|
||||
}
|
||||
}
|
||||
next.source?.let { pinSource(next.trackKey, it) }
|
||||
pinSources(next)
|
||||
}
|
||||
|
||||
private suspend fun pinSources(anchor: JamAnchor) {
|
||||
anchor.source?.let { pinSource(anchor.trackKey, it) }
|
||||
anchor.upcoming.forEach { pinSource(it.trackKey, it.source) }
|
||||
}
|
||||
|
||||
private suspend fun applyQueue(queue: JamQueue) {
|
||||
@ -610,6 +663,8 @@ class JamRoomService(
|
||||
if (items == lastAppliedItems) return
|
||||
val previous = lastAppliedItems
|
||||
lastAppliedItems = items
|
||||
// Sources that arrived before the queue had their tracks, before the queue loads them.
|
||||
anchor?.let { pinSources(it) }
|
||||
|
||||
// The host only appended items (e.g. accepted suggestions): merge them in
|
||||
// without disturbing playback.
|
||||
@ -643,9 +698,12 @@ class JamRoomService(
|
||||
if (now < holdSyncUntilMs) return SyncStatus.Busy
|
||||
|
||||
val current = audioPlayerQueue.currentQueueEntryFlow.value
|
||||
onEntryChanged(current?.jamKey())
|
||||
if (current?.jamKey() != room.trackKey) {
|
||||
val index = audioPlayerQueue.queueFlow.value.indexOfFirst { it.jamKey() == room.trackKey }
|
||||
val queue = audioPlayerQueue.queueFlow.value
|
||||
val index = queue.indexOfFirst { it.jamKey() == room.trackKey }
|
||||
if (index < 0) return SyncStatus.Idle // The queue with this track hasn't arrived yet.
|
||||
if (finishedSongFirst(queue, current, index, room, now)) return SyncStatus.Busy
|
||||
log.d { "Jam sync: switching to queue index $index" }
|
||||
holdSync(TRACK_CHANGE_HOLD_MS)
|
||||
corrector.reset()
|
||||
@ -700,34 +758,75 @@ class JamRoomService(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Our song ended a moment before the host's (players and recordings differ slightly in
|
||||
* length), so we're on the song after the room's. Wait for the host to get there
|
||||
* instead of going back to replay the ending. The room has to be near the end of its
|
||||
* song: a host that went back a song is followed right away.
|
||||
*/
|
||||
private fun finishedSongFirst(
|
||||
queue: List<QueueEntry>,
|
||||
current: QueueEntry?,
|
||||
roomIndex: Int,
|
||||
room: JamAnchor,
|
||||
now: Long): Boolean {
|
||||
if (!room.playing || current == null) return false
|
||||
if (queue.indexOfFirst { it.jamKey() == current.jamKey() } != roomIndex + 1) return false
|
||||
if (now - entryChangedAtMs > NEXT_SONG_WAIT_MS) return false
|
||||
val roomSongMs = roomQueue?.items?.firstOrNull { it.key == room.trackKey }?.durationMs
|
||||
?: (queue[roomIndex] as? QueueEntry.StreamingTrack)?.track?.durationMs
|
||||
?: return false
|
||||
return roomSongMs > 0 && room.positionAt(jamClient.clock.serverTime(now)) >= roomSongMs - NEAR_SONG_END_MS
|
||||
}
|
||||
|
||||
/**
|
||||
* Uses the host's audio source for [trackKey] when we use the same audio plugin, so
|
||||
* both sides stream the same recording. Uploads of one song differ in intro length,
|
||||
* which no amount of clock sync can fix.
|
||||
* which no amount of clock sync can fix. Upcoming songs are pinned well before our
|
||||
* player loads them; a song already playing from another recording is reloaded.
|
||||
*/
|
||||
private suspend fun pinSource(trackKey: String, tagged: PluginTaggedAudioSource) {
|
||||
if (!pinnedSources.add("$trackKey|${tagged.source.id}")) return
|
||||
val pin = "$trackKey|${tagged.source.id}"
|
||||
if (pin in pinnedSources) return
|
||||
val plugin = audioPluginSource.selectedAudioPlugin.value ?: return
|
||||
if (plugin.pluginId != tagged.pluginId) return
|
||||
val entry = audioPlayerQueue.queueFlow.value
|
||||
.firstOrNull { it is QueueEntry.StreamingTrack && it.track.id == trackKey } as? QueueEntry.StreamingTrack
|
||||
?: return
|
||||
val mine = runCatching { trackSources.getTrackSource(entry.track, plugin.pluginId) }.getOrNull()
|
||||
// Not in the room's queue yet: tried again when the queue arrives (applyQueue).
|
||||
val track = roomTrack(trackKey) ?: return
|
||||
pinnedSources += pin
|
||||
val mine = runCatching { trackSources.getTrackSource(track, plugin.pluginId) }.getOrNull()
|
||||
if (mine == tagged.source) return
|
||||
|
||||
log.i { "Jam: using the host's source ${tagged.source.id} for $trackKey" }
|
||||
runCatching {
|
||||
trackSources.saveTrackSource(entry.track, tagged.source, plugin.pluginId)
|
||||
trackSources.saveTrackSource(track, tagged.source, plugin.pluginId)
|
||||
streamingUrlRepository.invalidateCachedStreamUrl(trackKey)
|
||||
if (audioPlayerQueue.currentQueueEntryFlow.value?.jamKey() == trackKey) {
|
||||
syncMutex.withLock {
|
||||
holdSync(TRACK_CHANGE_HOLD_MS)
|
||||
audioPlayerQueue.reloadCurrent()
|
||||
}
|
||||
}
|
||||
// A cached copy of another recording would be played instead.
|
||||
evictCachedAudio(trackKey)
|
||||
if (audioPlayerQueue.currentQueueEntryFlow.value?.jamKey() == trackKey) reloadForNewSource(trackKey)
|
||||
}.onFailure { log.w(it) { "Failed to switch to the host's audio source" } }
|
||||
}
|
||||
|
||||
/** [trackKey]'s track, from our queue or (before it has loaded) the room's. */
|
||||
private fun roomTrack(trackKey: String): MetadataTrack? =
|
||||
audioPlayerQueue.queueFlow.value
|
||||
.firstNotNullOfOrNull { (it as? QueueEntry.StreamingTrack)?.track?.takeIf { t -> t.id == trackKey } }
|
||||
?: roomQueue?.items?.firstOrNull { it.trackId == trackKey }?.sanitized()
|
||||
?.let { (it.toQueueEntry() as? QueueEntry.StreamingTrack)?.track }
|
||||
|
||||
/**
|
||||
* Plays the current song again from the newly pinned source. Jumping to the song that's
|
||||
* playing only seeks within what's already loaded on Android (ExoPlayer), so the queue
|
||||
* is reloaded at this song instead.
|
||||
*/
|
||||
private suspend fun reloadForNewSource(trackKey: String) = syncMutex.withLock {
|
||||
val queue = audioPlayerQueue.queueFlow.value
|
||||
val index = queue.indexOfFirst { it.jamKey() == trackKey }.takeIf { it >= 0 } ?: return@withLock
|
||||
holdSync(TRACK_CHANGE_HOLD_MS)
|
||||
positionEntryKey = null
|
||||
audioPlayerQueue.load(queue, autoPlay = anchor?.playing ?: true, startPosition = index)
|
||||
onEntryChanged(trackKey)
|
||||
}
|
||||
|
||||
private suspend fun setRate(rate: Float) {
|
||||
val rounded = (rate * 200).toInt() / 200f
|
||||
if (appliedRate == rounded) return
|
||||
@ -750,6 +849,21 @@ class JamRoomService(
|
||||
|
||||
// ---------- Position tracking ----------
|
||||
|
||||
/**
|
||||
* A new song starts from zero. Without this, the old song's last position would carry
|
||||
* over: a song that ends on its own doesn't make the player report a jump like a skip
|
||||
* does, and its next report can still be the old song's.
|
||||
*/
|
||||
private fun onEntryChanged(key: String?) {
|
||||
if (key == positionEntryKey) return
|
||||
positionEntryKey = key
|
||||
entryChangedAtMs = now()
|
||||
positionSampleMs = 0
|
||||
positionSampledAtMs = entryChangedAtMs
|
||||
positionAdvancing = audioPlayer.playerStateFlow.value == PlayerState.PLAYING
|
||||
corrector.reset()
|
||||
}
|
||||
|
||||
/** The player's position now, extrapolated from its last (coarse) report. */
|
||||
private fun localPositionMs(): Long {
|
||||
if (!positionAdvancing) return positionSampleMs
|
||||
@ -766,6 +880,12 @@ class JamRoomService(
|
||||
|
||||
private fun onPosition(positionMs: Long) {
|
||||
val now = now()
|
||||
onEntryChanged(audioPlayerQueue.currentQueueEntryFlow.value?.jamKey())
|
||||
// Right after a song change, a report can still be the old song's position.
|
||||
val sinceChange = now - entryChangedAtMs
|
||||
if (sinceChange < TRACK_CHANGE_HOLD_MS &&
|
||||
positionMs > sinceChange * audioPlayer.playbackSpeedFlow.value + STALE_REPORT_SLACK_MS
|
||||
) return
|
||||
// A seek moves the position away from both the last report and the extrapolation;
|
||||
// a stall (or a long pause) only from the extrapolation.
|
||||
val jumped = abs(positionMs - positionSampleMs) > USER_SEEK_THRESHOLD_MS &&
|
||||
@ -876,6 +996,13 @@ class JamRoomService(
|
||||
private const val QUEUE_MIN_GAP_MS = 1_000L
|
||||
private const val ANCHOR_MIN_GAP_MS = 250L
|
||||
private const val MAX_SUGGESTED_ITEMS = 500
|
||||
/** How many next songs the host shares its sources for. */
|
||||
private const val UPCOMING_SOURCES = 2
|
||||
/** How long a guest that finished a song first waits for the host to finish it too. */
|
||||
private const val NEXT_SONG_WAIT_MS = 15_000L
|
||||
/** "Near the end" of the room's song, allowing for recordings of different lengths. */
|
||||
private const val NEAR_SONG_END_MS = 15_000L
|
||||
private const val STALE_REPORT_SLACK_MS = 1_500L
|
||||
private const val HOST_DRIFT_CHECK_INTERVAL_MS = 2_000L
|
||||
private const val HOST_DRIFT_WINDOW = 5
|
||||
/** Below the guests' dead band, so the host's own error never adds up to a correction. */
|
||||
|
||||
@ -40,6 +40,11 @@ import org.spotube.plugin_interfaces.plugin_apis.audio.AudioSource
|
||||
import org.spotube.plugin_interfaces.plugin_apis.metadata.track.MetadataTrack
|
||||
import org.spotube.spotube.core.audioplayer.AudioPlayerInterface
|
||||
import org.spotube.spotube.core.audioplayer.AudioPlayerQueue
|
||||
import org.spotube.spotube.core.audioplayer.DeviceAudioPlayerQueue
|
||||
import org.spotube.spotube.core.audioplayer.FakeAudioPlayerQueueRepository
|
||||
import org.spotube.spotube.core.audioplayer.FakeBlacklistRepository
|
||||
import org.spotube.spotube.core.audioplayer.FakePluginProvider
|
||||
import org.spotube.spotube.core.audioplayer.FakeSettingsProvider
|
||||
import org.spotube.spotube.core.audioplayer.LoopState
|
||||
import org.spotube.spotube.core.audioplayer.MediaItem
|
||||
import org.spotube.spotube.core.audioplayer.PlayerState
|
||||
@ -161,6 +166,176 @@ class JamSessionSimulationTest {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Songs that end on their own. The guest has its own uploads of every song (a shorter
|
||||
* intro) and its player loads the next song halfway through the current one, so it
|
||||
* must have the host's recording before then, or it plays a different one.
|
||||
*/
|
||||
@Test
|
||||
fun usesTheHostsRecordingForEverySong() = runBlocking {
|
||||
val url = serverUrl ?: return@runBlocking
|
||||
val http = HttpClientFactory()
|
||||
val tracks = List(3) { track("song-$it", durationMs = SONG_MS) }
|
||||
val host = Participant(http, url, speedError = 1.0, realQueue = true)
|
||||
val guest = Participant(http, url, speedError = 1.01, realQueue = true)
|
||||
tracks.forEach {
|
||||
host.sources.seed(it, source("host-${it.id}"), PLUGIN)
|
||||
guest.sources.seed(it, source("own-${it.id}"), PLUGIN)
|
||||
}
|
||||
try {
|
||||
host.queue.load(tracks.map { QueueEntry.StreamingTrack(it, url = "") }, autoPlay = true, startPosition = 0)
|
||||
delay(500)
|
||||
val code = host.service.createRoom().getOrThrow()
|
||||
guest.service.joinRoom(code).getOrThrow()
|
||||
delay(6_000)
|
||||
assertEquals("host-song-0", guest.player.currentSource, "guest plays the host's recording of song 0")
|
||||
assertInSync(host, guest, "during song 0")
|
||||
|
||||
for (song in 1..2) {
|
||||
while (host.player.index.value < song) delay(100)
|
||||
delay(5_000)
|
||||
assertEquals(song, guest.player.index.value, "guest on song $song")
|
||||
assertEquals("host-song-$song", guest.player.currentSource, "guest plays the host's recording of song $song")
|
||||
assertInSync(host, guest, "5 s into song $song after the previous one ended")
|
||||
}
|
||||
} finally {
|
||||
guest.close()
|
||||
host.close()
|
||||
http.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A guest that can't use the host's recordings (another audio plugin) finishes each
|
||||
* song early. It must wait for the host, not jump back to the song that just ended.
|
||||
*/
|
||||
@Test
|
||||
fun aGuestThatFinishesASongFirstWaitsForTheHost() = runBlocking {
|
||||
val url = serverUrl ?: return@runBlocking
|
||||
val http = HttpClientFactory()
|
||||
val tracks = List(3) { track("song-$it", durationMs = SONG_MS) }
|
||||
val host = Participant(http, url, speedError = 1.0)
|
||||
val guest = Participant(http, url, speedError = 1.0, pluginId = "other-plugin")
|
||||
tracks.forEach { guest.sources.seed(it, source("own-${it.id}"), "other-plugin") }
|
||||
val jumpedBack = CopyOnWriteArrayList<String>()
|
||||
try {
|
||||
host.queue.load(tracks.map { QueueEntry.StreamingTrack(it, url = "") }, autoPlay = true, startPosition = 0)
|
||||
delay(500)
|
||||
val code = host.service.createRoom().getOrThrow()
|
||||
guest.service.joinRoom(code).getOrThrow()
|
||||
delay(4_000)
|
||||
val watcher = launch {
|
||||
var last = guest.player.index.value
|
||||
guest.player.index.collect { now ->
|
||||
if (now < last) jumpedBack += "guest went back from song $last to $now"
|
||||
last = now
|
||||
}
|
||||
}
|
||||
for (song in 1..2) {
|
||||
while (host.player.index.value < song) delay(100)
|
||||
delay(5_000)
|
||||
assertEquals(song, guest.player.index.value, "guest on song $song")
|
||||
}
|
||||
watcher.cancel()
|
||||
assertTrue(jumpedBack.isEmpty(), jumpedBack.toString())
|
||||
} finally {
|
||||
guest.close()
|
||||
host.close()
|
||||
http.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A jam must not keep the CPU busy (a spinning loop allocates nonstop and freezes the
|
||||
* app in GC). Measures the process's CPU time while the room sits in each state.
|
||||
*/
|
||||
@Test
|
||||
fun idleStatesDoNotBurnCpu() = runBlocking {
|
||||
val url = serverUrl ?: return@runBlocking
|
||||
val http = HttpClientFactory()
|
||||
val tracks = List(3) { track("cpu-$it") }
|
||||
val host = Participant(http, url, speedError = 1.0, realQueue = true)
|
||||
val guest = Participant(http, url, speedError = 1.01, realQueue = true)
|
||||
val os = java.lang.management.ManagementFactory.getOperatingSystemMXBean() as com.sun.management.OperatingSystemMXBean
|
||||
suspend fun cpuDuring(label: String, block: suspend () -> Unit) {
|
||||
val before = os.processCpuTime
|
||||
val start = System.nanoTime()
|
||||
block()
|
||||
val share = (os.processCpuTime - before).toDouble() / (System.nanoTime() - start)
|
||||
println("Jam simulation: CPU ${"%.0f".format(share * 100)}% of a core while $label")
|
||||
assertTrue(share < 0.5, "CPU ${"%.0f".format(share * 100)}% of a core while $label")
|
||||
}
|
||||
try {
|
||||
host.queue.load(tracks.map { QueueEntry.StreamingTrack(it, url = "") }, autoPlay = true, startPosition = 0)
|
||||
delay(500)
|
||||
val code = host.service.createRoom().getOrThrow()
|
||||
guest.service.joinRoom(code).getOrThrow()
|
||||
delay(5_000)
|
||||
cpuDuring("playing in sync") { delay(8_000) }
|
||||
host.player.pause()
|
||||
delay(2_000)
|
||||
cpuDuring("the room is paused") { delay(8_000) }
|
||||
host.player.play()
|
||||
delay(3_000)
|
||||
host.player.stall()
|
||||
delay(1_000)
|
||||
cpuDuring("the host is buffering") { delay(8_000) }
|
||||
host.player.unstall()
|
||||
delay(3_000)
|
||||
guest.player.stall()
|
||||
cpuDuring("the guest is buffering") { delay(8_000) }
|
||||
guest.player.unstall()
|
||||
} finally {
|
||||
guest.close()
|
||||
host.close()
|
||||
http.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Several minutes of jam (songs ending, recordings switching, seeks) must not grow the
|
||||
* heap. Opt-in, since it takes minutes: set `JAM_SOAK=1`.
|
||||
*/
|
||||
@Test
|
||||
fun longSessionDoesNotLeak() = runBlocking {
|
||||
val url = serverUrl ?: return@runBlocking
|
||||
if (System.getenv("JAM_SOAK") == null) return@runBlocking
|
||||
val http = HttpClientFactory()
|
||||
val tracks = List(8) { track("soak-$it", durationMs = SONG_MS) }
|
||||
val host = Participant(http, url, speedError = 1.0, realQueue = true)
|
||||
val guest = Participant(http, url, speedError = 1.01, realQueue = true)
|
||||
tracks.forEach {
|
||||
host.sources.seed(it, source("host-${it.id}"), PLUGIN)
|
||||
guest.sources.seed(it, source("own-${it.id}"), PLUGIN)
|
||||
}
|
||||
val runtime = Runtime.getRuntime()
|
||||
fun usedMb(): Long {
|
||||
repeat(3) { System.gc(); Thread.sleep(200) }
|
||||
return (runtime.totalMemory() - runtime.freeMemory()) / (1024 * 1024)
|
||||
}
|
||||
try {
|
||||
host.queue.load(tracks.map { QueueEntry.StreamingTrack(it, url = "") }, autoPlay = true, startPosition = 0)
|
||||
delay(500)
|
||||
val code = host.service.createRoom().getOrThrow()
|
||||
guest.service.joinRoom(code).getOrThrow()
|
||||
delay(20_000)
|
||||
val baseline = usedMb()
|
||||
val samples = mutableListOf(baseline)
|
||||
repeat(6) { round ->
|
||||
if (round % 2 == 0) host.player.seekTo(10_000.milliseconds) else guest.player.seekTo(20_000.milliseconds)
|
||||
delay(30_000)
|
||||
samples += usedMb()
|
||||
}
|
||||
println("Jam simulation: heap after GC over ${samples.size} samples (MB): $samples")
|
||||
assertTrue(samples.last() - baseline < 30, "heap grew from $baseline MB to ${samples.last()} MB")
|
||||
assertInSync(host, guest, "after a long session")
|
||||
} finally {
|
||||
guest.close()
|
||||
host.close()
|
||||
http.close()
|
||||
}
|
||||
}
|
||||
|
||||
/** Losing the network (guest, then host) must not crash, and the room picks up again. */
|
||||
@Test
|
||||
fun roomSurvivesNetworkLoss() = runBlocking {
|
||||
@ -223,16 +398,32 @@ class JamSessionSimulationTest {
|
||||
|
||||
private fun assertInSync(host: Participant, guest: Participant, phase: String) {
|
||||
assertEquals(host.player.index.value, guest.player.index.value, "same track $phase")
|
||||
val gap = guest.player.truePosition() - host.player.truePosition()
|
||||
val gap = guest.player.songPosition() - host.player.songPosition()
|
||||
println("Jam simulation: guest is ${gap}ms off $phase")
|
||||
assertTrue(abs(gap) < MAX_GAP_MS, "guest is ${gap}ms off $phase")
|
||||
}
|
||||
|
||||
private class Participant(http: HttpClientFactory, serverUrl: String, speedError: Double) {
|
||||
val player = SimPlayer(speedError)
|
||||
val queue = SimQueue(player)
|
||||
private class Participant(
|
||||
http: HttpClientFactory,
|
||||
serverUrl: String,
|
||||
speedError: Double,
|
||||
pluginId: String = PLUGIN,
|
||||
/** The app's own queue (stateIn flows and all) instead of [SimQueue]. */
|
||||
realQueue: Boolean = false) {
|
||||
val sources = FakeTrackSourceRepository()
|
||||
private val plugins = FakeAudioPluginSource(FakeAudioPluginService(PLUGIN))
|
||||
val player = SimPlayer(speedError)
|
||||
val queue: AudioPlayerQueue = if (realQueue) {
|
||||
player.resolveSource = { i ->
|
||||
val id = player.playlistFlow.value.getOrNull(i)?.url?.substringAfterLast('/')
|
||||
id?.let { sources.rawEntryFor(it)?.source?.id } ?: "default-$id"
|
||||
}
|
||||
DeviceAudioPlayerQueue(
|
||||
player, FakeSettingsProvider(), FakeAudioPlayerQueueRepository(),
|
||||
FakePluginProvider(), FakeBlacklistRepository())
|
||||
} else {
|
||||
SimQueue(player, sources)
|
||||
}
|
||||
private val plugins = FakeAudioPluginSource(FakeAudioPluginService(pluginId))
|
||||
val service = JamRoomService(
|
||||
jamClient = JamRoomClient(http),
|
||||
audioPlayer = player,
|
||||
@ -241,7 +432,8 @@ class JamSessionSimulationTest {
|
||||
audioPluginSource = plugins,
|
||||
trackSources = sources,
|
||||
streamingUrlRepository = StreamingUrlRepository(
|
||||
plugins, FakeStreamingAudioPlayerQueue(), sources, FakeUserSettingsSource()))
|
||||
plugins, FakeStreamingAudioPlayerQueue(), sources, FakeUserSettingsSource()),
|
||||
evictCachedAudio = {})
|
||||
|
||||
suspend fun close() {
|
||||
service.leaveRoom()
|
||||
@ -260,10 +452,18 @@ class JamSessionSimulationTest {
|
||||
* Plays a list of tracks in real time. Its clock runs [speedError] times true speed (as
|
||||
* a drifting device would), positions are reported every 250 ms rounded to 100 ms, and
|
||||
* every seek or track change buffers briefly.
|
||||
*
|
||||
* Like ExoPlayer, it picks a song's recording (source) when it loads the song: the next
|
||||
* song is loaded halfway through the current one, and jumping to the song already
|
||||
* playing keeps its recording. Recordings differ in intro length ([introMs]).
|
||||
*/
|
||||
private class SimPlayer(private val speedError: Double) : AudioPlayerInterface {
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
private var durations: List<Long> = emptyList()
|
||||
/** The source the queue would resolve for a song now. */
|
||||
var resolveSource: (Int) -> String = { "default" }
|
||||
private val loadedSources = mutableMapOf<Int, String>()
|
||||
val currentSource: String? get() = loadedSources[index.value]
|
||||
private var basePosition = 0L
|
||||
private var baseAt = now()
|
||||
private var speed = 1f
|
||||
@ -274,6 +474,8 @@ class JamSessionSimulationTest {
|
||||
private val _position = MutableStateFlow(Duration.ZERO)
|
||||
private val _duration = MutableStateFlow(Duration.ZERO)
|
||||
private val _speed = MutableStateFlow(1f)
|
||||
private val _playlist = MutableStateFlow<List<MediaItem>>(emptyList())
|
||||
private val _currentItem = MutableStateFlow<MediaItem?>(null)
|
||||
private val _completion = MutableSharedFlow<Unit>(extraBufferCapacity = 4)
|
||||
|
||||
init {
|
||||
@ -281,10 +483,14 @@ class JamSessionSimulationTest {
|
||||
while (isActive) {
|
||||
delay(250)
|
||||
val position = truePosition()
|
||||
val duration = durations.getOrNull(index.value) ?: continue
|
||||
val duration = currentDuration() ?: continue
|
||||
val next = index.value + 1
|
||||
if (position >= duration / 2 && next <= durations.lastIndex && next !in loadedSources) {
|
||||
loadedSources[next] = resolveSource(next)
|
||||
}
|
||||
if (position >= duration) {
|
||||
_completion.tryEmit(Unit)
|
||||
if (index.value < durations.lastIndex) startTrack(index.value + 1)
|
||||
if (index.value < durations.lastIndex) advanceGapless()
|
||||
continue
|
||||
}
|
||||
_position.value = (position / 100 * 100).milliseconds
|
||||
@ -292,6 +498,12 @@ class JamSessionSimulationTest {
|
||||
}
|
||||
}
|
||||
|
||||
/** Where in the song (not the recording) playback is. */
|
||||
fun songPosition(): Long = truePosition() - introMs(currentSource)
|
||||
|
||||
private fun currentDuration(): Long? =
|
||||
durations.getOrNull(index.value)?.let { it + introMs(currentSource) }
|
||||
|
||||
@Synchronized
|
||||
fun truePosition(): Long =
|
||||
if (_state.value == PlayerState.PLAYING) {
|
||||
@ -309,12 +521,30 @@ class JamSessionSimulationTest {
|
||||
fun loadTracks(trackDurations: List<Long>, start: Int, autoPlay: Boolean) {
|
||||
durations = trackDurations
|
||||
playWhenReady = autoPlay
|
||||
loadedSources.clear()
|
||||
startTrack(start)
|
||||
}
|
||||
|
||||
/**
|
||||
* A song ending on its own, like ExoPlayer's gapless transition: no buffering, and
|
||||
* the position keeps the old song's value until the next report.
|
||||
*/
|
||||
private fun advanceGapless() {
|
||||
val next = index.value + 1
|
||||
loadedSources.getOrPut(next) { resolveSource(next) }
|
||||
index.value = next
|
||||
_currentItem.value = _playlist.value.getOrNull(next)
|
||||
_duration.value = currentDuration()!!.milliseconds
|
||||
rebase(0)
|
||||
}
|
||||
|
||||
fun startTrack(i: Int) {
|
||||
// Jumping to the song already playing seeks within it (ExoPlayer's
|
||||
// seekToDefaultPosition): the loaded recording stays.
|
||||
if (i != index.value || i !in loadedSources) loadedSources[i] = loadedSources[i] ?: resolveSource(i)
|
||||
index.value = i
|
||||
_duration.value = durations[i].milliseconds
|
||||
_currentItem.value = _playlist.value.getOrNull(i)
|
||||
_duration.value = currentDuration()!!.milliseconds
|
||||
rebase(0)
|
||||
_position.value = Duration.ZERO
|
||||
buffer()
|
||||
@ -330,6 +560,17 @@ class JamSessionSimulationTest {
|
||||
}
|
||||
}
|
||||
|
||||
/** Network stall: buffering until [unstall]. */
|
||||
fun stall() {
|
||||
rebase()
|
||||
_state.value = PlayerState.BUFFERING
|
||||
}
|
||||
|
||||
fun unstall() {
|
||||
rebase()
|
||||
_state.value = if (playWhenReady) PlayerState.PLAYING else PlayerState.PAUSED
|
||||
}
|
||||
|
||||
override suspend fun play() {
|
||||
playWhenReady = true
|
||||
rebase()
|
||||
@ -364,8 +605,8 @@ class JamSessionSimulationTest {
|
||||
override val playbackSpeedFlow: StateFlow<Float> = _speed
|
||||
override val completionFlow: Flow<Unit> = _completion
|
||||
override val shuffleModeFlow: StateFlow<Boolean> = MutableStateFlow(false)
|
||||
override val currentMediaItemFlow: StateFlow<MediaItem?> = MutableStateFlow(null)
|
||||
override val playlistFlow: StateFlow<List<MediaItem>> = MutableStateFlow(emptyList())
|
||||
override val currentMediaItemFlow: StateFlow<MediaItem?> = _currentItem
|
||||
override val playlistFlow: StateFlow<List<MediaItem>> = _playlist
|
||||
override val bufferingPositionFlow: StateFlow<Duration> = MutableStateFlow(Duration.ZERO)
|
||||
override val loopStateFlow: StateFlow<LoopState> = MutableStateFlow(LoopState.NONE)
|
||||
override val volumeFlow: StateFlow<Float> = MutableStateFlow(1f)
|
||||
@ -374,8 +615,15 @@ class JamSessionSimulationTest {
|
||||
override suspend fun stop() = pause()
|
||||
override suspend fun loop(state: LoopState) = Unit
|
||||
override suspend fun shuffle(enabled: Boolean) = Unit
|
||||
override suspend fun load(playlist: List<MediaItem>, autoPlay: Boolean, startPosition: Int) = error("use SimQueue")
|
||||
override suspend fun addMediaItem(mediaItem: MediaItem) = Unit
|
||||
override suspend fun load(playlist: List<MediaItem>, autoPlay: Boolean, startPosition: Int) {
|
||||
_playlist.value = playlist
|
||||
loadTracks(playlist.map { it.duration.inWholeMilliseconds }, startPosition, autoPlay)
|
||||
}
|
||||
|
||||
override suspend fun addMediaItem(mediaItem: MediaItem) {
|
||||
_playlist.value += mediaItem
|
||||
durations = durations + mediaItem.duration.inWholeMilliseconds
|
||||
}
|
||||
override suspend fun insertMediaItemAtNextIndex(mediaItem: MediaItem) = Unit
|
||||
override suspend fun removeMediaItem(mediaItem: MediaItem) = Unit
|
||||
override suspend fun moveMediaItem(fromIndex: Int, toIndex: Int) = Unit
|
||||
@ -384,10 +632,17 @@ class JamSessionSimulationTest {
|
||||
override fun dispose() = scope.cancel()
|
||||
}
|
||||
|
||||
private class SimQueue(private val player: SimPlayer) : AudioPlayerQueue {
|
||||
private class SimQueue(private val player: SimPlayer, sources: FakeTrackSourceRepository) : AudioPlayerQueue {
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
private val entries = MutableStateFlow<List<QueueEntry>>(emptyList())
|
||||
|
||||
init {
|
||||
player.resolveSource = { i ->
|
||||
val id = (entries.value.getOrNull(i) as? QueueEntry.StreamingTrack)?.track?.id
|
||||
id?.let { sources.rawEntryFor(it)?.source?.id } ?: "default-$id"
|
||||
}
|
||||
}
|
||||
|
||||
override val queueFlow: StateFlow<List<QueueEntry>> = entries
|
||||
override val currentQueueEntryFlow: StateFlow<QueueEntry?> =
|
||||
combine(entries, player.index) { list, i -> list.getOrNull(i) }
|
||||
@ -431,11 +686,15 @@ class JamSessionSimulationTest {
|
||||
private const val BUFFER_MS = 200L
|
||||
private const val MAX_GAP_MS = 200L
|
||||
private const val STEADY_WINDOW_MS = 40_000L
|
||||
private const val SONG_MS = 30_000L
|
||||
|
||||
/** How much longer a recording's intro is than the original's ("own" uploads cut 2.5 s). */
|
||||
private fun introMs(source: String?): Long = if (source?.startsWith("own-") == true) -2_500 else 0
|
||||
|
||||
private fun now() = Clock.System.now().toEpochMilliseconds()
|
||||
|
||||
private fun track(id: String) = MetadataTrack(
|
||||
id = id, title = id, durationMs = 180_000, trackNumber = null, discNumber = null,
|
||||
private fun track(id: String, durationMs: Long = 180_000) = MetadataTrack(
|
||||
id = id, title = id, durationMs = durationMs, trackNumber = null, discNumber = null,
|
||||
artists = emptyList(), album = null, thumbnails = null, explicit = null,
|
||||
popularity = null, isrcCode = null, externalUri = null)
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user