diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/AudioPlayerQueue.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/AudioPlayerQueue.kt index 51d47129..96989f17 100644 --- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/AudioPlayerQueue.kt +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/AudioPlayerQueue.kt @@ -44,6 +44,12 @@ interface AudioPlayerQueue { suspend fun getCurrentCollectionEntry(): QueueCollectionEntry? suspend fun getCollectionHistory(): List + /** + * 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 diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/DeviceAudioPlayerQueue.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/DeviceAudioPlayerQueue.kt index fb283055..ee7f5542 100644 --- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/DeviceAudioPlayerQueue.kt +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/audioplayer/DeviceAudioPlayerQueue.kt @@ -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> = 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 = 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 = 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 diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/di/Modules.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/di/Modules.kt index 4c45a3fd..98d487df 100644 --- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/di/Modules.kt +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/di/Modules.kt @@ -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()::invalidateCacheEntry) + } singleOf(::AudioPlayerQueueRepository) { bind() } single { DeviceAudioPlayerQueue(get(), get(), get(), get(), get()) diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt index 7744c5f7..195e3fe3 100644 --- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt @@ -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 = emptyList()) { /** The room's position at server time [serverNowMs]. */ fun positionAt(serverNowMs: Long): Long = if (advancing) positionMs + ((serverNowMs - serverTimeMs) * rate).toLong() else positionMs diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt index 5f5bd986..90ea8e6b 100644 --- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt @@ -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 = emptyList() private val anchorRequests = Channel(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() + .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, + 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. */ diff --git a/composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/jam/JamSessionSimulationTest.kt b/composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/jam/JamSessionSimulationTest.kt index 88c33db8..1bba4a76 100644 --- a/composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/jam/JamSessionSimulationTest.kt +++ b/composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/jam/JamSessionSimulationTest.kt @@ -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() + 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 = emptyList() + /** The source the queue would resolve for a song now. */ + var resolveSource: (Int) -> String = { "default" } + private val loadedSources = mutableMapOf() + 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>(emptyList()) + private val _currentItem = MutableStateFlow(null) private val _completion = MutableSharedFlow(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, 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 = _speed override val completionFlow: Flow = _completion override val shuffleModeFlow: StateFlow = MutableStateFlow(false) - override val currentMediaItemFlow: StateFlow = MutableStateFlow(null) - override val playlistFlow: StateFlow> = MutableStateFlow(emptyList()) + override val currentMediaItemFlow: StateFlow = _currentItem + override val playlistFlow: StateFlow> = _playlist override val bufferingPositionFlow: StateFlow = MutableStateFlow(Duration.ZERO) override val loopStateFlow: StateFlow = MutableStateFlow(LoopState.NONE) override val volumeFlow: StateFlow = 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, autoPlay: Boolean, startPosition: Int) = error("use SimQueue") - override suspend fun addMediaItem(mediaItem: MediaItem) = Unit + override suspend fun load(playlist: List, 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>(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> = entries override val currentQueueEntryFlow: StateFlow = 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)