diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b0203208f2..167220374e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ * Bug Fixes * Allow copying description and show notes links with a long press ([#5628](https://github.com/Automattic/pocket-casts-android/pull/5628)) + * Prevent simultaneous audio streams on Android Automotive + ([#5636](https://github.com/Automattic/pocket-casts-android/pull/5636)) 8.17 ----- diff --git a/modules/services/repositories/build.gradle.kts b/modules/services/repositories/build.gradle.kts index 03b765f57df..d5987306b37 100644 --- a/modules/services/repositories/build.gradle.kts +++ b/modules/services/repositories/build.gradle.kts @@ -82,6 +82,7 @@ dependencies { testImplementation(libs.okHttp.mockwebserver) testImplementation(libs.robolectric) testImplementation(libs.turbine) + testImplementation(libs.work.test) testImplementation(projects.modules.services.sharedtest) testImplementation(projects.modules.services.analytics.testing) diff --git a/modules/services/repositories/src/main/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlaybackManager.kt b/modules/services/repositories/src/main/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlaybackManager.kt index f17df8a3a34..3b46f3c0b82 100644 --- a/modules/services/repositories/src/main/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlaybackManager.kt +++ b/modules/services/repositories/src/main/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlaybackManager.kt @@ -123,6 +123,7 @@ import kotlin.time.Duration.Companion.milliseconds import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.asStateFlow @@ -224,10 +225,13 @@ open class PlaybackManager @Inject constructor( val playbackStateFlow: Flow = playbackStateRelay.asFlow() private var updateCount = 0 + + @Volatile private var resettingPlayer = false + + private val playerTransitions = PlayerTransitionCoordinator() private var episodeLastBufferStatus: EpisodeBufferStatus? = null private var focusWasPlaying: Date? = null - private var forcePlayerSwitch = false private var updateTimerDisposable: Disposable? = null private var bufferUpdateTimerDisposable: Disposable? = null private var pauseTimerDisposable: Disposable? = null @@ -365,7 +369,12 @@ open class PlaybackManager @Inject constructor( return currentEpisode.podcastUuid.let { podcastManager.findPodcastByUuidBlocking(it) } } - private suspend fun autoLoadEpisode(autoPlay: Boolean): BaseEpisode? { + private suspend fun autoLoadEpisode( + autoPlay: Boolean, + transitionVersion: PlayerTransitionCoordinator.Token, + ): BaseEpisode? { + if (!playerTransitions.isCurrent(transitionVersion)) return null + val nextEpisode = getCurrentEpisode() if (nextEpisode != null) { return nextEpisode @@ -377,14 +386,20 @@ open class PlaybackManager @Inject constructor( // auto queue next episode on empty val autoPlayEpisode = autoSelectNextEpisode() ?: return null + if (!playerTransitions.isCurrent(transitionVersion)) return null - withContext(Dispatchers.Default) { - upNextQueue.playNextBlocking(autoPlayEpisode) { - launch { - loadCurrentEpisode(play = autoPlay, sourceView = SourceView.AUTO_PLAY) - } + val episodeQueued = playerTransitions.runIfCurrent(transitionVersion) { + withContext(Dispatchers.Default) { + upNextQueue.playNextBlocking(autoPlayEpisode, onAdd = null) } } + if (!episodeQueued) return null + + loadCurrentEpisode( + play = autoPlay, + sourceView = SourceView.AUTO_PLAY, + transitionVersion = transitionVersion, + ) eventHorizon.track( PlaybackEpisodeAutoplayedEvent( @@ -399,11 +414,70 @@ open class PlaybackManager @Inject constructor( return playbackStateRelay.blockingFirst().isPlaying } - private suspend fun applyStreamOverride(episode: BaseEpisode) { + private suspend fun withTransitionCompletion( + transitionVersion: PlayerTransitionCoordinator.Token, + block: suspend () -> T, + ): T { + try { + return block() + } finally { + withContext(NonCancellable) { + playerTransitions.completeTransition(transitionVersion) { + eventSourceForCompletedTransition(transitionVersion) + } + } + } + } + + private suspend fun eventSourceForCompletedTransition( + transitionVersion: PlayerTransitionCoordinator.Token, + ): Player? { + val currentPlayer = player ?: return null + val currentEpisodeUuid = getCurrentEpisode()?.uuid + val playerEpisodeUuid = currentPlayer.episodeUuid + val hasEventSource = playerTransitions.hasEventSource(currentPlayer) + if (hasEventSource && playerEpisodeUuid == currentEpisodeUuid) { + return currentPlayer + } + if (playerTransitions.hasInactiveEventSourceFromEarlierTransition(currentPlayer, transitionVersion)) { + return currentPlayer + } + + if (player === currentPlayer) { + stopPlayback() + } + return null + } + + private fun launchTransition( + transitionVersion: PlayerTransitionCoordinator.Token, + context: CoroutineContext = coroutineContext, + block: suspend () -> Unit, + ): Job { + return launch(context) { + withTransitionCompletion(transitionVersion, block) + } + } + + private suspend fun withOptionalTransitionCompletion( + transitionVersion: PlayerTransitionCoordinator.Token?, + block: suspend () -> T, + ): T { + return if (transitionVersion == null) { + block() + } else { + withTransitionCompletion(transitionVersion, block) + } + } + + private suspend fun resolveStreamOverride(episode: BaseEpisode): String? { + return alternateEnclosureManager.findForEpisode(episode.uuid).firstHlsStreamUrl() + } + + private fun applyStreamOverride(episode: BaseEpisode, hlsUrl: String?) { episode.overrideStreamUrl = null episode.overrideStreamContentType = null // Stream the first HLS alternate enclosure when streaming is on, or when the episode is HLS-only. - val hlsUrl = alternateEnclosureManager.findForEpisode(episode.uuid).firstHlsStreamUrl() _streamHlsAvailable.value = hlsUrl != null val hlsStreamingEnabled = FeatureFlag.isEnabled(Feature.HLS_STREAMING) if (hlsUrl != null && (hlsStreamingEnabled || episode.downloadUrl.isNullOrBlank())) { @@ -464,16 +538,19 @@ open class PlaybackManager @Inject constructor( isVideoToggleReloading.set(false) return } - launch(Dispatchers.Default) { + val transitionVersion = playerTransitions.beginTransition() + val play = isPlaying() + launchTransition(transitionVersion, Dispatchers.Default) { try { player?.let { player -> val currentTimeSecs = player.getCurrentPositionMs().toDouble() / 1000.0 episodeManager.updatePlayedUpToBlocking(episode, currentTimeSecs, true) } loadCurrentEpisode( - play = isPlaying(), + play = play, forceStream = streamWarningConfirmed, showedStreamWarning = streamWarningConfirmed, + transitionVersion = transitionVersion, ) } finally { isVideoToggleReloading.set(false) @@ -625,15 +702,29 @@ open class PlaybackManager @Inject constructor( sourceView: SourceView = SourceView.UNKNOWN, showedStreamWarning: Boolean = false, ) { - launch { playQueueSuspend(sourceView, showedStreamWarning) } + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { + playQueueSuspend(sourceView, showedStreamWarning, transitionVersion) + } } suspend fun playQueueSuspend( sourceView: SourceView = SourceView.UNKNOWN, showedStreamWarning: Boolean = false, + ) { + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + playQueueSuspend(sourceView, showedStreamWarning, transitionVersion) + } + } + + private suspend fun playQueueSuspend( + sourceView: SourceView, + showedStreamWarning: Boolean, + transitionVersion: PlayerTransitionCoordinator.Token, ) { if (upNextQueue.currentEpisode != null) { - loadEpisodeWhenRequired(sourceView, showedStreamWarning) + loadEpisodeWhenRequired(sourceView, showedStreamWarning, transitionVersion) } } @@ -643,12 +734,14 @@ open class PlaybackManager @Inject constructor( showedStreamWarning: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN, ) { - launch { + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { playNowSuspend( episode = episode, forceStream = forceStream, showedStreamWarning = showedStreamWarning, sourceView = sourceView, + transitionVersion = transitionVersion, ) } } @@ -659,8 +752,13 @@ open class PlaybackManager @Inject constructor( showedStreamWarning: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN, ) { - val episode = episodeManager.findEpisodeByUuid(episodeUuid) ?: return - playNowSuspend(episode, forceStream, showedStreamWarning, sourceView) + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + val episode = episodeManager.findEpisodeByUuid(episodeUuid) + if (episode != null) { + playNowSuspend(episode, forceStream, showedStreamWarning, sourceView, transitionVersion) + } + } } suspend fun playNowSuspend( @@ -669,12 +767,32 @@ open class PlaybackManager @Inject constructor( showedStreamWarning: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN, ) { - forcePlayerSwitch = true + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + playNowSuspend( + episode = episode, + forceStream = forceStream, + showedStreamWarning = showedStreamWarning, + sourceView = sourceView, + transitionVersion = transitionVersion, + ) + } + } + + private suspend fun playNowSuspend( + episode: BaseEpisode, + forceStream: Boolean, + showedStreamWarning: Boolean, + sourceView: SourceView, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { playNowSync( episode = episode, forceStream = forceStream, showedStreamWarning = showedStreamWarning, sourceView = sourceView, + transitionVersion = transitionVersion, + forcePlayerSwitch = true, ) } @@ -683,6 +801,27 @@ open class PlaybackManager @Inject constructor( forceStream: Boolean = false, showedStreamWarning: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN, + ) { + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + playNowSync( + episode = episode, + forceStream = forceStream, + showedStreamWarning = showedStreamWarning, + sourceView = sourceView, + transitionVersion = transitionVersion, + forcePlayerSwitch = false, + ) + } + } + + private suspend fun playNowSync( + episode: BaseEpisode, + forceStream: Boolean, + showedStreamWarning: Boolean, + sourceView: SourceView, + transitionVersion: PlayerTransitionCoordinator.Token, + forcePlayerSwitch: Boolean, ) { LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Play now: ${episode.uuid} ${episode.title}") @@ -696,29 +835,36 @@ open class PlaybackManager @Inject constructor( } } + if (!playerTransitions.isCurrent(transitionVersion)) return + val switchEpisode: Boolean = !upNextQueue.isCurrentEpisode(episode) - if (switchEpisode || isPlayerSwitchRequired()) { + if (switchEpisode || isPlayerSwitchRequired(forcePlayerSwitch)) { LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Player switch required. Different episode: $switchEpisode") - pause(transientLoss = true) - upNextQueue.playNow( - episode = episode, - automaticUpNextSource = autoPlaySource(sourceView, episode), - onAdd = { - launch { - loadCurrentEpisode( - play = true, - forceStream = forceStream, - showedStreamWarning = showedStreamWarning, - sourceView = sourceView, - ) - } - }, + val episodeQueued = playerTransitions.runIfCurrent(transitionVersion) { + player?.let(playerTransitions::clearEventSource) + pausePlayer(transientLoss = true, sourceView = SourceView.UNKNOWN) + upNextQueue.playNow( + episode = episode, + automaticUpNextSource = autoPlaySource(sourceView, episode), + onAdd = null, + ) + } + if (!episodeQueued) return + + loadCurrentEpisode( + play = true, + forceStream = forceStream, + showedStreamWarning = showedStreamWarning, + sourceView = sourceView, + forcePlayerReset = forcePlayerSwitch && switchEpisode, + transitionVersion = transitionVersion, ) } else if (playbackStateRelay.blockingFirst().isPaused) { LogBuffer.i(LogBuffer.TAG_PLAYBACK, "No player switch required. Playing queue.") - playQueue( + playQueueSuspend( showedStreamWarning = showedStreamWarning, sourceView = sourceView, + transitionVersion = transitionVersion, ) } } @@ -820,21 +966,26 @@ open class PlaybackManager @Inject constructor( episode: BaseEpisode, source: SourceView, userInitiated: Boolean = true, - ) = withContext(Dispatchers.Default) { + ) { val wasEmpty: Boolean = upNextQueue.isEmpty - upNextQueue.playNextBlocking(episode, onAdd = null, isUserInitiated = userInitiated) - if (userInitiated) { - eventHorizon.track( - EpisodeAddedToUpNextEvent( - episodeUuid = episode.uuid, - toTop = true, - source = source.analyticsValue, - ), - ) - notificationManager.updateUserFeatureInteraction(OnboardingNotificationType.UpNext) - } - if (wasEmpty) { - loadCurrentEpisode(play = false) + val transitionVersion = if (wasEmpty) playerTransitions.beginTransition() else null + withOptionalTransitionCompletion(transitionVersion) { + withContext(Dispatchers.Default) { + upNextQueue.playNextBlocking(episode, onAdd = null, isUserInitiated = userInitiated) + if (userInitiated) { + eventHorizon.track( + EpisodeAddedToUpNextEvent( + episodeUuid = episode.uuid, + toTop = true, + source = source.analyticsValue, + ), + ) + notificationManager.updateUserFeatureInteraction(OnboardingNotificationType.UpNext) + } + if (transitionVersion != null) { + loadCurrentEpisode(play = false, transitionVersion = transitionVersion) + } + } } } @@ -844,34 +995,42 @@ open class PlaybackManager @Inject constructor( userInitiated: Boolean = true, ) { val wasEmpty: Boolean = upNextQueue.isEmpty - upNextQueue.playLast(episode, onAdd = null, isUserInitiated = userInitiated) - if (userInitiated) { - eventHorizon.track( - EpisodeAddedToUpNextEvent( - episodeUuid = episode.uuid, - toTop = false, - source = source.analyticsValue, - ), - ) - notificationManager.updateUserFeatureInteraction(OnboardingNotificationType.UpNext) - } - if (wasEmpty) { - loadCurrentEpisode(play = false) + val transitionVersion = if (wasEmpty) playerTransitions.beginTransition() else null + withOptionalTransitionCompletion(transitionVersion) { + upNextQueue.playLast(episode, onAdd = null, isUserInitiated = userInitiated) + if (userInitiated) { + eventHorizon.track( + EpisodeAddedToUpNextEvent( + episodeUuid = episode.uuid, + toTop = false, + source = source.analyticsValue, + ), + ) + notificationManager.updateUserFeatureInteraction(OnboardingNotificationType.UpNext) + } + if (transitionVersion != null) { + loadCurrentEpisode(play = false, transitionVersion = transitionVersion) + } } } private suspend fun loadEpisodeWhenRequired( - sourceView: SourceView = SourceView.UNKNOWN, - showedStreamWarning: Boolean = false, + sourceView: SourceView, + showedStreamWarning: Boolean, + transitionVersion: PlayerTransitionCoordinator.Token, ) { if (isPlayerSwitchRequired()) { loadCurrentEpisode( play = true, showedStreamWarning = showedStreamWarning, sourceView = sourceView, + transitionVersion = transitionVersion, ) } else { - play(sourceView) + playerTransitions.runIfCurrent(transitionVersion) { + player?.let { playerTransitions.bindEventSource(it, transitionVersion) } + play(sourceView) + } } } @@ -904,11 +1063,23 @@ open class PlaybackManager @Inject constructor( return } - launch { + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { val topEpisode = episodes.first() - playNowSync(episode = topEpisode, sourceView = sourceView) + playNowSync( + episode = topEpisode, + forceStream = false, + showedStreamWarning = false, + sourceView = sourceView, + transitionVersion = transitionVersion, + forcePlayerSwitch = false, + ) if (episodes.size > 1) { - upNextQueue.clearAndPlayAll(episodes.slice(1 until min(episodes.size, settings.getMaxUpNextEpisodes()))) + playerTransitions.runIfCurrent(transitionVersion) { + upNextQueue.clearAndPlayAll( + episodes.slice(1 until min(episodes.size, settings.getMaxUpNextEpisodes())), + ) + } } } } @@ -918,11 +1089,14 @@ open class PlaybackManager @Inject constructor( return } + val wasEmpty = upNextQueue.isEmpty + val transitionVersion = if (wasEmpty) playerTransitions.beginTransition() else null launch { - val wasEmpty = upNextQueue.isEmpty - addEpisodesLast(episodes, source) - if (wasEmpty) { - loadCurrentEpisode(play = false) + withOptionalTransitionCompletion(transitionVersion) { + addEpisodesLast(episodes, source) + if (transitionVersion != null) { + loadCurrentEpisode(play = false, transitionVersion = transitionVersion) + } } } } @@ -948,20 +1122,23 @@ open class PlaybackManager @Inject constructor( return } + val wasEmpty = upNextQueue.isEmpty + val transitionVersion = if (wasEmpty) playerTransitions.beginTransition() else null launch { - val currentEpisode = upNextQueue.currentEpisode?.uuid - val wasEmpty: Boolean = upNextQueue.isEmpty - upNextQueue.playAllNext(episodes.filter { it.uuid != currentEpisode }) - eventHorizon.track( - EpisodeBulkAddToUpNextEvent( - count = episodes.size.toLong(), - toTop = true, - source = source.analyticsValue, - ), - ) - notificationManager.updateUserFeatureInteraction(OnboardingNotificationType.UpNext) - if (wasEmpty) { - loadCurrentEpisode(play = false) + withOptionalTransitionCompletion(transitionVersion) { + val currentEpisode = upNextQueue.currentEpisode?.uuid + upNextQueue.playAllNext(episodes.filter { it.uuid != currentEpisode }) + eventHorizon.track( + EpisodeBulkAddToUpNextEvent( + count = episodes.size.toLong(), + toTop = true, + source = source.analyticsValue, + ), + ) + notificationManager.updateUserFeatureInteraction(OnboardingNotificationType.UpNext) + if (transitionVersion != null) { + loadCurrentEpisode(play = false, transitionVersion = transitionVersion) + } } } } @@ -973,12 +1150,31 @@ open class PlaybackManager @Inject constructor( } fun pause(transientLoss: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN) { - launch { - pauseSuspend(transientLoss, sourceView) + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { + pauseSuspend(transientLoss, sourceView, transitionVersion) } } suspend fun pauseSuspend(transientLoss: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN) { + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + pauseSuspend(transientLoss, sourceView, transitionVersion) + } + } + + private suspend fun pauseSuspend( + transientLoss: Boolean, + sourceView: SourceView, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { + playerTransitions.runIfCurrent(transitionVersion) { + player?.let { playerTransitions.bindEventSource(it, transitionVersion) } + pausePlayer(transientLoss, sourceView) + } + } + + private suspend fun pausePlayer(transientLoss: Boolean, sourceView: SourceView) { if (!transientLoss) { focusManager.giveUpAudioFocus() playbackStateRelay.blockingFirst().let { playbackState -> @@ -1004,12 +1200,30 @@ open class PlaybackManager @Inject constructor( } fun stopAsync(isAudioFocusFailed: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN) { - launch { - stopSuspend(isAudioFocusFailed, sourceView) + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { + stopSuspend(isAudioFocusFailed, sourceView, transitionVersion) } } suspend fun stopSuspend(isAudioFocusFailed: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN) { + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + stopSuspend(isAudioFocusFailed, sourceView, transitionVersion) + } + } + + private suspend fun stopSuspend( + isAudioFocusFailed: Boolean, + sourceView: SourceView, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { + playerTransitions.runIfCurrent(transitionVersion) { + stopPlayback(isAudioFocusFailed, sourceView) + } + } + + private suspend fun stopPlayback(isAudioFocusFailed: Boolean, sourceView: SourceView) { if (!isAudioFocusFailed) { trackPlaybackEvent(sourceView) { source, contentType -> PlaybackStopEvent( @@ -1018,10 +1232,23 @@ open class PlaybackManager @Inject constructor( ) } } - stop() + stopPlayback() } suspend fun stop() { + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + stop(transitionVersion) + } + } + + private suspend fun stop(transitionVersion: PlayerTransitionCoordinator.Token) { + playerTransitions.runIfCurrent(transitionVersion) { + stopPlayback() + } + } + + private suspend fun stopPlayback() { LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Stopping playback") cancelPrefetchNextEpisode() @@ -1029,9 +1256,15 @@ open class PlaybackManager @Inject constructor( cancelBufferUpdateTimer() withContext(Dispatchers.Main) { - if (player != null) { - player?.stop() - player = null + player?.let { playerToStop -> + playerTransitions.clearEventSource(playerToStop) + try { + playerToStop.stop() + } catch (exception: Exception) { + Timber.e(exception, "Failed to stop player") + } finally { + player = null + } } } @@ -1047,15 +1280,33 @@ open class PlaybackManager @Inject constructor( } suspend fun shutdown() { - stop() + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + shutdown(transitionVersion) + } + } - audioNoisyManager.unregister() - focusManager.giveUpAudioFocus() + private suspend fun shutdown( + transitionVersion: PlayerTransitionCoordinator.Token, + beforeCastSessionEnd: suspend () -> Unit = {}, + ) { + playerTransitions.runIfCurrent(transitionVersion) { + stopPlayback() - withContext(Dispatchers.Main) { - playbackStateRelay.accept(PlaybackState(state = PlaybackState.State.EMPTY, lastChangeFrom = LastChangeFrom.OnShutdown.value)) + audioNoisyManager.unregister() + focusManager.giveUpAudioFocus() + + withContext(Dispatchers.Main) { + playbackStateRelay.accept( + PlaybackState( + state = PlaybackState.State.EMPTY, + lastChangeFrom = LastChangeFrom.OnShutdown.value, + ), + ) + } + beforeCastSessionEnd() + castManager.endSession() } - castManager.endSession() } suspend fun hibernatePlayback() { @@ -1135,32 +1386,55 @@ open class PlaybackManager @Inject constructor( sourceView: SourceView = SourceView.UNKNOWN, jumpAmountSeconds: Int = settings.skipForwardInSecs.value, ) { - launch { - skipForwardSuspend(sourceView, jumpAmountSeconds) + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { + skipForwardSuspend(sourceView, jumpAmountSeconds, transitionVersion) } } suspend fun skipForwardSuspend( sourceView: SourceView = SourceView.UNKNOWN, jumpAmountSeconds: Int = settings.skipForwardInSecs.value, + ) { + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + skipForwardSuspend(sourceView, jumpAmountSeconds, transitionVersion) + } + } + + private suspend fun skipForwardSuspend( + sourceView: SourceView, + jumpAmountSeconds: Int, + transitionVersion: PlayerTransitionCoordinator.Token, ) { LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Skip forward tapped") - val episode = getCurrentEpisode() ?: return + val episode = getCurrentEpisode() + if (episode == null) { + return + } val jumpAmountMs = jumpAmountSeconds * 1000 val currentTimeMs = getCurrentTimeMs(episode = episode) - if (currentTimeMs < 0 || player?.episodeUuid != episode.uuid) return // Make sure the player hasn't changed episodes before using the current time to seek + if (!playerTransitions.isCurrent(transitionVersion)) return + if (currentTimeMs < 0 || player?.episodeUuid != episode.uuid) { + // Make sure the player hasn't changed episodes before using the current time to seek. + return + } val newPositionMs = currentTimeMs + jumpAmountMs val durationMs = player?.durationMs() ?: Int.MAX_VALUE // If we don't have a duration, just let them skip - statsManager.addTimeSavedSkipping((newPositionMs - currentTimeMs).toLong()) if (newPositionMs < durationMs) { - seekToTimeMsInternal(newPositionMs) + playerTransitions.runIfCurrent(transitionVersion) { + player?.let { playerTransitions.bindEventSource(it, transitionVersion) } + statsManager.addTimeSavedSkipping((newPositionMs - currentTimeMs).toLong()) + seekToTimeMsInternal(newPositionMs) + } } else { + statsManager.addTimeSavedSkipping((newPositionMs - currentTimeMs).toLong()) LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Seek beyond end of episode so completing. ${episode.uuid}") - onCompletion(episode.uuid) + player?.let { onPlayerEvent(it, PlayerEvent.Completion(episode.uuid)) } } trackPlaybackEvent(sourceView) { source, contentType -> @@ -1275,10 +1549,12 @@ open class PlaybackManager @Inject constructor( } fun endPlaybackAndClearUpNextAsync() { - launch { - shutdown() - upNextHistoryManager.snapshotUpNext() - upNextQueue.removeAll() + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { + shutdown(transitionVersion) { + upNextHistoryManager.snapshotUpNext() + upNextQueue.removeAll() + } } } @@ -1304,96 +1580,215 @@ open class PlaybackManager @Inject constructor( private val removeMutex = Mutex() fun removeEpisode(episodeToRemove: BaseEpisode?, source: SourceView, userInitiated: Boolean = true, shouldShuffleUpNext: Boolean = false) { - launch { - if (episodeToRemove == null) { - return@launch - } - - removeMutex.withLock { - val currentEpisode = getCurrentEpisode() + removeEpisodeAsync(episodeToRemove, source, userInitiated, shouldShuffleUpNext) + } - val isCurrentEpisode = currentEpisode != null && currentEpisode.uuid == episodeToRemove.uuid && (player == null || player?.episodeUuid == episodeToRemove.uuid) - val isPlaying = isPlaying() + internal fun removeEpisodeAsync( + episodeToRemove: BaseEpisode?, + source: SourceView, + userInitiated: Boolean = true, + shouldShuffleUpNext: Boolean = false, + ): Job? { + if (episodeToRemove == null) return null - if (isCurrentEpisode) { - // when there is another episode in the Up Next and we are playing, don't stop so the foreground service isn't stopped - val moreEpisodes = upNextQueue.size > 1 - if (moreEpisodes && isPlaying) { - pause(transientLoss = true) - } else { - stop() - } - } + return launch { + removeEpisodeDurably( + episodeToRemove = episodeToRemove, + source = source, + userInitiated = userInitiated, + shouldShuffleUpNext = shouldShuffleUpNext, + ) + } + } - upNextQueue.removeEpisode(episodeToRemove, shouldShuffleUpNext) - if (userInitiated) { - eventHorizon.track( - EpisodeRemovedFromUpNextEvent( - episodeUuid = episodeToRemove.uuid, - source = source.analyticsValue, - ), + private suspend fun removeEpisodeDurably( + episodeToRemove: BaseEpisode, + source: SourceView, + userInitiated: Boolean, + shouldShuffleUpNext: Boolean, + ) { + while (true) { + val settledSnapshot = playerTransitions.awaitSettledSnapshot() + val transitionVersion = playerTransitions.tryBeginTransition(settledSnapshot) ?: continue + val removal = withTransitionCompletion(transitionVersion) { + val result = removeMutex.withLock { + removeEpisodeForTransition( + episodeToRemove = episodeToRemove, + source = source, + userInitiated = userInitiated, + shouldShuffleUpNext = shouldShuffleUpNext, + transitionVersion = transitionVersion, ) } + loadEpisodeAfterRemoval(result, transitionVersion) + result + } + if (removal.isRemoved) return + } + } + + private data class EpisodeRemovalResult( + val isRemoved: Boolean = false, + val shouldLoadNextEpisode: Boolean = false, + val shouldPlayNextEpisode: Boolean = false, + ) - if (isCurrentEpisode) { - loadCurrentEpisode(play = isPlaying, sourceView = SourceView.AUTO_PLAY) + private suspend fun removeEpisodeForTransition( + episodeToRemove: BaseEpisode, + source: SourceView, + userInitiated: Boolean, + shouldShuffleUpNext: Boolean, + transitionVersion: PlayerTransitionCoordinator.Token, + ): EpisodeRemovalResult { + var removal = EpisodeRemovalResult() + val episodeRemoved = playerTransitions.runIfCurrent(transitionVersion) { + val currentEpisodeUuid = getCurrentEpisode()?.uuid + val currentPlayerUuid = player?.episodeUuid + val removesCurrentPlayback = currentEpisodeUuid == episodeToRemove.uuid + val removesLoadedPlayer = currentPlayerUuid == episodeToRemove.uuid + val shouldLoadNextEpisode = removesCurrentPlayback || removesLoadedPlayer + val shouldPlayNextEpisode = shouldLoadNextEpisode && isPlaying() + + if (shouldLoadNextEpisode && currentPlayerUuid == episodeToRemove.uuid) { + // When there is another episode in Up Next and playback is active, pause instead of stopping so the + // foreground service remains alive while the next episode is loaded. + if (upNextQueue.size > 1 && shouldPlayNextEpisode) { + player?.let(playerTransitions::clearEventSource) + pausePlayer(transientLoss = true, sourceView = SourceView.UNKNOWN) + } else { + stopPlayback() } } + removeEpisodeFromQueue(episodeToRemove, source, userInitiated, shouldShuffleUpNext) + removal = EpisodeRemovalResult( + isRemoved = true, + shouldLoadNextEpisode = shouldLoadNextEpisode, + shouldPlayNextEpisode = shouldPlayNextEpisode, + ) + } + return removal.takeIf { episodeRemoved } ?: EpisodeRemovalResult() + } + + private suspend fun loadEpisodeAfterRemoval( + removal: EpisodeRemovalResult, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { + if (removal.shouldLoadNextEpisode) { + loadCurrentEpisode( + play = removal.shouldPlayNextEpisode, + sourceView = SourceView.AUTO_PLAY, + transitionVersion = transitionVersion, + ) + } + } + + private suspend fun removeEpisodeFromQueue( + episode: BaseEpisode, + source: SourceView, + userInitiated: Boolean, + shouldShuffleUpNext: Boolean, + ) { + upNextQueue.removeEpisode(episode, shouldShuffleUpNext) + trackEpisodeRemoved(episode, source, userInitiated) + } + + private fun trackEpisodeRemoved(episode: BaseEpisode, source: SourceView, userInitiated: Boolean) { + if (userInitiated) { + eventHorizon.track( + EpisodeRemovedFromUpNextEvent( + episodeUuid = episode.uuid, + source = source.analyticsValue, + ), + ) } } - private suspend fun onRemoteMetaDataNotMatched(episodeUuid: String) { + private suspend fun onRemoteMetaDataNotMatched( + episodeUuid: String, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { + if (!playerTransitions.isCurrent(transitionVersion)) return + val episode = episodeManager.findEpisodeByUuid(episodeUuid) ?: return - val podcast = if (episode is PodcastEpisode) podcastManager.findPodcastByUuidBlocking(episode.podcastUuid) else null + val podcast = if (episode is PodcastEpisode) { + podcastManager.findPodcastByUuidBlocking(episode.podcastUuid) + } else { + null + } + if (!playerTransitions.isCurrent(transitionVersion)) return if (player?.isRemote == true && player?.isPlaying() == false) { if (castManager.isPlaying()) { Timber.d("Playing remote episode %s", episode.title) - playNowSync(episode) + playNowSync( + episode = episode, + forceStream = false, + showedStreamWarning = false, + sourceView = SourceView.UNKNOWN, + transitionVersion = transitionVersion, + forcePlayerSwitch = false, + ) } else { - loadCurrentEpisode(play = false) + loadCurrentEpisode(play = false, transitionVersion = transitionVersion) + playerTransitions.runIfCurrent(transitionVersion) { + player?.let { playerTransitions.bindEventSource(it, transitionVersion) } + player?.setEpisode(episode) + player?.setPodcast(podcast) + } } - - player?.setEpisode(episode) - player?.setPodcast(podcast) } } fun castReconnected() { - launch(Dispatchers.Main) { - if (player != null) { - player?.stop() - player = null - } + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion, Dispatchers.Main) { + playerTransitions.runIfCurrent(transitionVersion) { + player?.let { playerToStop -> + playerTransitions.clearEventSource(playerToStop) + playerToStop.stop() + player = null + } - player = playerManager.createCastPlayer(this@PlaybackManager::onPlayerEvent) - mediaSessionManager.installCastPlayer() - (player as? CastPlayer)?.updateFromRemoteIfRequired() - Timber.i("Cast reconnected. Creating media player of type CastPlayer") + player = playerManager.createCastPlayer(this@PlaybackManager::onPlayerEvent) + player?.let { playerTransitions.bindEventSource(it, transitionVersion) } + mediaSessionManager.installCastPlayer() + (player as? CastPlayer)?.updateFromRemoteIfRequired() + Timber.i("Cast reconnected. Creating media player of type CastPlayer") - setupUpdateTimer() + setupUpdateTimer() + } } } fun castConnected() { upNextQueue.currentEpisode ?: return + val transitionVersion = playerTransitions.beginTransition() cancelPrefetchNextEpisode() - launch { + launchTransition(transitionVersion) { if (isPlayerSwitchRequired()) { - loadCurrentEpisode(true, sourceView = SourceView.CHROMECAST) + loadCurrentEpisode( + play = true, + sourceView = SourceView.CHROMECAST, + transitionVersion = transitionVersion, + ) } } } fun castDisconnected() { upNextQueue.currentEpisode ?: return - launch { + val transitionVersion = playerTransitions.beginTransition() + launchTransition(transitionVersion) { updateCurrentPositionInDatabase() - stop() + stop(transitionVersion) if (isPlayerSwitchRequired()) { - loadCurrentEpisode(false, sourceView = SourceView.CHROMECAST) + loadCurrentEpisode( + play = false, + sourceView = SourceView.CHROMECAST, + transitionVersion = transitionVersion, + ) } } } @@ -1402,6 +1797,19 @@ open class PlaybackManager @Inject constructor( @OptIn(UnstableApi::class) suspend fun onPlayerError(event: PlayerEvent.PlayerError) { + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + onPlayerError(event, transitionVersion) + } + } + + @OptIn(UnstableApi::class) + private suspend fun onPlayerError( + event: PlayerEvent.PlayerError, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { + if (!playerTransitions.isCurrent(transitionVersion)) return + settings.recordErrorSession() val episode = getCurrentEpisode() @@ -1438,28 +1846,45 @@ open class PlaybackManager @Inject constructor( } LogBuffer.e(LogBuffer.TAG_PLAYBACK, "Player error %s", event.message) + if (!playerTransitions.isCurrent(transitionVersion)) return // If a downloaded episode's file is missing, clear the stale download status // and retry playback via streaming instead of leaving the user stuck. if (episode != null && episode.isDownloaded && episode.downloadUrl != null && isDownloadedFileMissing(event)) { LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Downloaded file missing for ${episode.uuid}, clearing download status and retrying via stream") downloadQueue.cancel(episode.uuid, SourceView.UNKNOWN).join() + if (!playerTransitions.isCurrent(transitionVersion)) return + episodeManager.clearPlaybackErrorBlocking(episode) - playNow(episode = episode, forceStream = true, sourceView = SourceView.UNKNOWN) + playNowSuspend( + episode = episode, + forceStream = true, + showedStreamWarning = false, + sourceView = SourceView.UNKNOWN, + transitionVersion = transitionVersion, + ) return } val currentEpisode = getCurrentEpisode() - if (currentEpisode is BaseEpisode) { + if (!playerTransitions.isCurrent(transitionVersion)) return + if (currentEpisode?.uuid != episode?.uuid) { + return + } + if (currentEpisode != null) { episodeManager.markAsPlaybackErrorBlocking(currentEpisode, event, isPlaybackRemote()) } - stop() + stop(transitionVersion) + if (!playerTransitions.isCurrent(transitionVersion)) return + onPlayerPaused() val stuckException = event.error?.cause as? StuckPlayerException withContext(Dispatchers.Main) { + if (!playerTransitions.isCurrent(transitionVersion)) return@withContext + playbackStateRelay.blockingFirst().let { playbackState -> val cause = event.error?.cause val playbackIssue = when { @@ -1633,48 +2058,107 @@ open class PlaybackManager @Inject constructor( mediaSessionManager.updateCastState(isPlaying, isBuffering, positionMs) } - private suspend fun onCompletion(episodeUUID: String?) { + private class CompletionTransitionState( + val episodeUuid: String?, + var episode: BaseEpisode? = null, + var isInitialized: Boolean = false, + var isIgnored: Boolean = false, + var isPrepared: Boolean = false, + var isEpisodeRemoved: Boolean = false, + var isEpisodeFinalized: Boolean = false, + var shouldAutoPlay: Boolean = false, + ) + + private suspend fun onCompletion( + state: CompletionTransitionState, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { + if (!playerTransitions.isCurrent(transitionVersion)) return if (resettingPlayer) { return } - // keep hold of this as deleting the episode might change the member variable - val hadSleepAfterEpisode = isSleepAfterEpisodeEnabled() - val wasPlaying = isPlaying() + if (!state.isInitialized) { + val episode = getCurrentEpisode() + if (episode != null && episode.uuid != state.episodeUuid) { + // We have already completed this episode, don't do it again or we may skip the next one. + LogBuffer.e( + LogBuffer.TAG_PLAYBACK, + "OnCompletion uuid does not match playback state current episode, ignoring onComplete event.", + ) + state.isIgnored = true + } else { + state.episode = episode + } + state.isInitialized = true + } + if (state.isIgnored) return + + val episode = state.episode - cancelUpdateTimer() - cancelBufferUpdateTimer() + if (!state.isPrepared) { + // Keep hold of these as deleting the episode might change the member variables. + val hadSleepAfterEpisode = isSleepAfterEpisodeEnabled() + val wasPlaying = isPlaying() - val episode = getCurrentEpisode() + cancelUpdateTimer() + cancelBufferUpdateTimer() + + LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Episode ${episode?.title} finished, should sleep: $hadSleepAfterEpisode") - LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Episode ${episode?.title} finished, should sleep: $hadSleepAfterEpisode") + if (hadSleepAfterEpisode) { + sleepEndOfEpisode(episode) + if (!isSleepAfterEpisodeEnabled()) { + state.isIgnored = true + return + } + } - if (hadSleepAfterEpisode) { - sleepEndOfEpisode(episode) - if (!isSleepAfterEpisodeEnabled()) return + cancelUpdateTimer() + cancelBufferUpdateTimer() + + if (episode != null) { + eventHorizon.track( + PlayerEpisodeCompletedEvent( + podcastUuid = episode.podcastOrSubstituteUuid, + episodeUuid = episode.uuid, + hlsAvailable = _streamHlsAvailable.value, + audioOnlyMode = audioOnlyModeOrNull(), + ), + ) + } + + state.shouldAutoPlay = (!hadSleepAfterEpisode || isSleepAfterEpisodeEnabled()) && wasPlaying + state.isPrepared = true } - cancelUpdateTimer() - cancelBufferUpdateTimer() + if (!playerTransitions.isCurrent(transitionVersion)) return - if (episode != null) { - if (episode.uuid != episodeUUID) { - // We have already completed this episode, don't do it again or we may skip the next one - LogBuffer.e(LogBuffer.TAG_PLAYBACK, "OnCompletion uuid does not match playback state current episode, ignoring onComplete event.") + if (episode != null && !state.isEpisodeRemoved) { + var episodeRemoved = false + val transitionCommitted = playerTransitions.runIfCurrent(transitionVersion) { + when { + upNextQueue.currentEpisode?.uuid == episode.uuid -> { + upNextQueue.removeEpisode(episode, shouldShuffleUpNext = settings.upNextShuffle.value) + episodeRemoved = true + state.isEpisodeRemoved = true + } + + player?.episodeUuid == episode.uuid -> { + // A prior attempt committed the queue removal before a newer command superseded its player load. + episodeRemoved = true + state.isEpisodeRemoved = true + } + } + } + if (!transitionCommitted) return + if (!episodeRemoved) { + state.isIgnored = true return } - eventHorizon.track( - PlayerEpisodeCompletedEvent( - podcastUuid = episode.podcastOrSubstituteUuid, - episodeUuid = episode.uuid, - hlsAvailable = _streamHlsAvailable.value, - audioOnlyMode = audioOnlyModeOrNull(), - ), - ) - - // remove from Up Next - upNextQueue.removeEpisode(episode, shouldShuffleUpNext = settings.upNextShuffle.value) + } + if (episode != null && !state.isEpisodeFinalized) { // mark as played episodeManager.updatePlayingStatusBlocking(episode, EpisodePlayingStatus.COMPLETED) @@ -1716,23 +2200,25 @@ open class PlaybackManager @Inject constructor( } } } + state.isEpisodeFinalized = true } - // Auto play if it had sleep time enabled for end of episodes and still has episodes set on sleep time - // or if it did not have sleep time end of episode configured - // and it was playing episode - val autoPlay = (!hadSleepAfterEpisode || isSleepAfterEpisodeEnabled()) && wasPlaying + if (!playerTransitions.isCurrent(transitionVersion)) return var nextEpisode = getCurrentEpisode() if (nextEpisode == null) { - nextEpisode = autoLoadEpisode(autoPlay) + nextEpisode = autoLoadEpisode(state.shouldAutoPlay && isPlaying(), transitionVersion) if (nextEpisode == null) { + if (!playerTransitions.isCurrent(transitionVersion)) return lastTrackedAutoPlaySource = null - stop() - shutdown() + shutdown(transitionVersion) } } else { - loadCurrentEpisode(play = autoPlay, sourceView = SourceView.AUTO_PLAY) + loadCurrentEpisode( + play = state.shouldAutoPlay && isPlaying(), + sourceView = SourceView.AUTO_PLAY, + transitionVersion = transitionVersion, + ) } } @@ -1762,24 +2248,34 @@ open class PlaybackManager @Inject constructor( if (episode == null) return sleepTimer.sleepEndOfEpisode(episode) { - showToast(application.getString(LR.string.player_sleep_time_fired)) + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) transition@{ + showToast(application.getString(LR.string.player_sleep_time_fired)) + if (!playerTransitions.isCurrent(transitionVersion)) return@transition + + val podcast = playbackStateRelay.blockingFirst().podcast + if (podcast != null && podcast.skipLastSecs > 0) { + pauseSuspend( + transientLoss = false, + sourceView = SourceView.AUTO_PAUSE, + transitionVersion = transitionVersion, + ) + if (!playerTransitions.isCurrent(transitionVersion)) return@transition - val podcast = playbackStateRelay.blockingFirst().podcast - if (podcast != null && podcast.skipLastSecs > 0) { - pause(sourceView = SourceView.AUTO_PAUSE) - onPlayerPaused() - } + onPlayerPaused() + } - // jump back 5 seconds from the current time so when the player opens it doesn't complete before giving the user a chance to skip back - player?.let { - val currentTimeMs = it.getCurrentPositionMs() - 5000 - if (currentTimeMs > 0) { - val currentTimeSecs = currentTimeMs.toDouble() / 1000.0 - episodeManager.updatePlayedUpToBlocking(episode, currentTimeSecs, false) + // jump back 5 seconds from the current time so when the player opens it doesn't complete before giving the user a chance to skip back + player?.let { + val currentTimeMs = it.getCurrentPositionMs() - 5000 + if (currentTimeMs > 0) { + val currentTimeSecs = currentTimeMs.toDouble() / 1000.0 + episodeManager.updatePlayedUpToBlocking(episode, currentTimeSecs, false) + } } - } - stop() + stop(transitionVersion) + } } } @@ -1981,12 +2477,11 @@ open class PlaybackManager @Inject constructor( /** * Check the player is initialised and if we are using the correct player either the system or cast player. */ - private suspend fun isPlayerSwitchRequired(): Boolean { + private suspend fun isPlayerSwitchRequired(forcePlayerSwitch: Boolean = false): Boolean { if (player == null) { return true } if (forcePlayerSwitch) { - forcePlayerSwitch = false return true } // using Chrome Cast make sure the player is connected @@ -2023,6 +2518,8 @@ open class PlaybackManager @Inject constructor( showedStreamWarning: Boolean = false, forceStream: Boolean = false, sourceView: SourceView = SourceView.UNKNOWN, + forcePlayerReset: Boolean = false, + transitionVersion: PlayerTransitionCoordinator.Token, ) { // make sure we have the most recent copy from the database val episode = when (val currentUpNextEpisode = upNextQueue.currentEpisode) { @@ -2048,23 +2545,20 @@ open class PlaybackManager @Inject constructor( } if (episode == null) { - val nextEpisode = autoLoadEpisode(autoPlay = play) + val nextEpisode = autoLoadEpisode(autoPlay = play, transitionVersion = transitionVersion) if (nextEpisode == null) { Timber.d("Playback: No episode in upnext, shutting down") - shutdown() + shutdown(transitionVersion) } return } + if (!playerTransitions.isCurrent(transitionVersion)) return + if (upNextQueue.currentEpisode?.uuid != episode.uuid) { + return + } val podcast = findPodcastByEpisode(episode) - cancelPauseTimer() - cancelUpdateTimer() - cancelBufferUpdateTimer() - - val currentPlayer = this.player - val sameEpisode = currentPlayer != null && episode.uuid == currentPlayer.episodeUuid - // completed episodes should play from the start if (episode.isFinished) { episodeManager.markAsNotPlayedBlocking(episode) @@ -2088,16 +2582,75 @@ open class PlaybackManager @Inject constructor( val newDownloadUrl = userEpisodeManager.getPlaybackUrlRxSingle(episode).await() episode.downloadUrl = newDownloadUrl } catch (e: Exception) { - onPlayerError(PlayerEvent.PlayerError("Could not load cloud file ${e.message}")) - removeEpisode(episode, source = sourceView) + if (!playerTransitions.isCurrent(transitionVersion)) { + return + } + if (upNextQueue.currentEpisode?.uuid != episode.uuid) { + return + } + + onPlayerError( + event = PlayerEvent.PlayerError("Could not load cloud file ${e.message}"), + transitionVersion = transitionVersion, + ) + if (!playerTransitions.isCurrent(transitionVersion)) return + + val removal = removeMutex.withLock { + removeEpisodeForTransition( + episodeToRemove = episode, + source = sourceView, + userInitiated = true, + shouldShuffleUpNext = false, + transitionVersion = transitionVersion, + ) + } + loadEpisodeAfterRemoval(removal, transitionVersion) return } } } } - // Resolve the HLS alternate enclosure so streamUrl reflects the stream that will play. - applyStreamOverride(episode) + val hlsStreamUrl = resolveStreamOverride(episode) + + playerTransitions.runIfCurrent(transitionVersion) { + val isCurrentEpisode = upNextQueue.currentEpisode?.uuid == episode.uuid + if (!isCurrentEpisode) { + LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Ignoring superseded episode load for ${episode.uuid}") + return@runIfCurrent + } + + // Apply the resolved enclosure only after confirming that this request is still current. + applyStreamOverride(episode, hlsStreamUrl) + loadCurrentEpisodeIntoPlayer( + episode = episode, + podcast = podcast, + play = play, + showedStreamWarning = showedStreamWarning, + forceStream = forceStream, + sourceView = sourceView, + forcePlayerReset = forcePlayerReset, + transitionVersion = transitionVersion, + ) + } + } + + private suspend fun loadCurrentEpisodeIntoPlayer( + episode: BaseEpisode, + podcast: Podcast?, + play: Boolean, + showedStreamWarning: Boolean, + forceStream: Boolean, + sourceView: SourceView, + forcePlayerReset: Boolean, + transitionVersion: PlayerTransitionCoordinator.Token, + ) { + cancelPauseTimer() + cancelUpdateTimer() + cancelBufferUpdateTimer() + + val currentPlayer = this.player + val sameEpisode = currentPlayer != null && episode.uuid == currentPlayer.episodeUuid if (videoStreamPreferred && episode.uuid != videoStreamPreferredEpisodeUuid) { videoStreamPreferred = false @@ -2177,12 +2730,14 @@ open class PlaybackManager @Inject constructor( if (player?.isStreaming == true && it.isDownloaded && player?.isRemote == false && !watchingVideo) { LogBuffer.i(LogBuffer.TAG_PLAYBACK, "Episode was streaming but was now downloaded, switching to downloaded file") - launch(Dispatchers.Default) { + val transitionVersion = playerTransitions.beginTransition() + val play = isPlaying() + launchTransition(transitionVersion, Dispatchers.Default) { player?.let { player -> val currentTimeSecs = player.getCurrentPositionMs().toDouble() / 1000.0 episodeManager.updatePlayedUpToBlocking(it, currentTimeSecs, true) } - loadCurrentEpisode(isPlaying()) + loadCurrentEpisode(play = play, transitionVersion = transitionVersion) } } else { Timber.d("Episode is not downloaded $this") @@ -2214,7 +2769,7 @@ open class PlaybackManager @Inject constructor( var posUpdatedOnPlayerReset = false // We want to make sure we get the current position at the last possible moment before changing/resetting the player val currentPositionMs = if ( - isPlayerSwitchRequired() || + isPlayerSwitchRequired(forcePlayerReset) || isPlayerResetNeeded(episode, sameEpisode, castManager.isConnected(), playingStream) ) { // Don't create a player if we aren't playing because it will start to buffer @@ -2244,8 +2799,16 @@ open class PlaybackManager @Inject constructor( player?.getCurrentPositionMs() } - player?.setPodcast(podcast) - player?.setEpisode(episode, videoStreamPreferred) + player?.let { + playerTransitions.bindEventSource(it, transitionVersion) + try { + it.setPodcast(podcast) + it.setEpisode(episode, videoStreamPreferred) + } catch (exception: Exception) { + playerTransitions.clearEventSource(it) + throw exception + } + } val playbackEffects = if (podcast != null && podcast.overrideGlobalEffects) { podcast.playbackEffects @@ -2484,53 +3047,114 @@ open class PlaybackManager @Inject constructor( } private suspend fun resetPlayer() { - if (resettingPlayer) return resettingPlayer = true - - withContext(Dispatchers.Main) { - player?.stop() - if (castManager.isConnected()) { - player = playerManager.createCastPlayer(this@PlaybackManager::onPlayerEvent) - mediaSessionManager.installCastPlayer() - Timber.i("Creating media player of type CastPlayer.") - } else { - player = playerManager.createSimplePlayer(this@PlaybackManager::onPlayerEvent) - // Start the service early so it's ready when we install the player later. - // The ExoPlayer doesn't exist yet — SimplePlayer creates it lazily in prepare(). - mediaSessionManager.startServiceIfNeeded(application) - Timber.i("Creating media player of type SimplePlayer.") + try { + withContext(Dispatchers.Main) { + player?.let { playerToStop -> + playerTransitions.clearEventSource(playerToStop) + playerToStop.stop() + } + if (castManager.isConnected()) { + player = playerManager.createCastPlayer(this@PlaybackManager::onPlayerEvent) + mediaSessionManager.installCastPlayer() + Timber.i("Creating media player of type CastPlayer.") + } else { + player = playerManager.createSimplePlayer(this@PlaybackManager::onPlayerEvent) + // Start the service early so it's ready when we install the player later. + // The ExoPlayer doesn't exist yet — SimplePlayer creates it lazily in prepare(). + mediaSessionManager.startServiceIfNeeded(application) + Timber.i("Creating media player of type SimplePlayer.") + } } + } finally { + resettingPlayer = false } - - resettingPlayer = false } private suspend fun stopPlayer() { withContext(Dispatchers.Main) { - player?.stop() + player?.let { playerToStop -> + playerTransitions.clearEventSource(playerToStop) + playerToStop.stop() + } } } private fun onPlayerEvent(player: Player, event: PlayerEvent) { - if (this.player != player) return + if (this.player !== player) return + val eventSourceToken = playerTransitions.tokenForEventSource(player) ?: return launch { Timber.d("Player %s event %s", player, event) when (event) { - is PlayerEvent.Completion -> onCompletion(event.episodeUUID) - is PlayerEvent.PlayerPaused -> onPlayerPaused() - is PlayerEvent.PlayerPlaying -> onPlayerPlaying() - is PlayerEvent.BufferingStateChanged -> onBufferingStateChanged() - is PlayerEvent.DurationAvailable -> onDurationAvailable() - is PlayerEvent.SeekComplete -> onSeekComplete(event.positionMs) - is PlayerEvent.MetadataAvailable -> onMetadataAvailable(event.metaData) - is PlayerEvent.PlayerError -> onPlayerError(event) - is PlayerEvent.RemoteMetadataNotMatched -> onRemoteMetaDataNotMatched(event.remoteEpisodeUuid) - is PlayerEvent.EpisodeChanged -> onEpisodeChanged(event.episodeUuid) - is PlayerEvent.CachingComplete -> onCachingComplete(event.episodeUuid) - is PlayerEvent.CachingReset -> onCachingReset(event.episodeUuid) - is PlayerEvent.VideoTrackChanged -> onVideoTrackChanged(event.hasVideo) + is PlayerEvent.Completion -> { + val completionState = CompletionTransitionState(event.episodeUUID) + handleTerminalPlayerEvent(player, eventSourceToken) { transitionVersion -> + onCompletion(completionState, transitionVersion) + } + } + + is PlayerEvent.PlayerError -> { + handleTerminalPlayerEvent(player, eventSourceToken) { transitionVersion -> + onPlayerError(event, transitionVersion) + } + } + + is PlayerEvent.RemoteMetadataNotMatched -> { + handleTerminalPlayerEvent(player, eventSourceToken) { transitionVersion -> + onRemoteMetaDataNotMatched(event.remoteEpisodeUuid, transitionVersion) + } + } + + else -> { + playerTransitions.runIfEventSourceCurrent(eventSourceToken) { + if (this@PlaybackManager.player !== player) return@runIfEventSourceCurrent + when (event) { + is PlayerEvent.PlayerPaused -> onPlayerPaused() + + is PlayerEvent.PlayerPlaying -> onPlayerPlaying() + + is PlayerEvent.BufferingStateChanged -> onBufferingStateChanged() + + is PlayerEvent.DurationAvailable -> onDurationAvailable() + + is PlayerEvent.SeekComplete -> onSeekComplete(event.positionMs) + + is PlayerEvent.MetadataAvailable -> onMetadataAvailable(event.metaData) + + is PlayerEvent.EpisodeChanged -> onEpisodeChanged(event.episodeUuid) + + is PlayerEvent.CachingComplete -> onCachingComplete(event.episodeUuid) + + is PlayerEvent.CachingReset -> onCachingReset(event.episodeUuid) + + is PlayerEvent.VideoTrackChanged -> onVideoTrackChanged(event.hasVideo) + + is PlayerEvent.Completion, + is PlayerEvent.PlayerError, + is PlayerEvent.RemoteMetadataNotMatched, + -> error("Terminal player events must be handled before passive events") + } + } + } + } + } + } + + private suspend fun handleTerminalPlayerEvent( + source: Player, + eventSourceToken: PlayerTransitionCoordinator.EventSourceToken, + handler: suspend (PlayerTransitionCoordinator.Token) -> Unit, + ) { + var transition = playerTransitions.beginTransitionForEventSource(eventSourceToken) ?: return + while (true) { + withTransitionCompletion(transition.token) terminalEvent@{ + if (player !== source || !playerTransitions.isCurrent(transition.token)) { + return@terminalEvent + } + handler(transition.token) } + transition = playerTransitions.retryTransitionForEventSource(transition.eventSourceToken) ?: return } } @@ -2664,17 +3288,26 @@ open class PlaybackManager @Inject constructor( currentChapterUuid = getCurrentChapterUuid(), currentEpisodeUuid = getCurrentEpisode()?.uuid, onSleepEndOfChapter = { - showToast(application.getString(LR.string.player_sleep_time_fired_end_of_chapter)) + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) transition@{ + showToast(application.getString(LR.string.player_sleep_time_fired_end_of_chapter)) + if (!playerTransitions.isCurrent(transitionVersion)) return@transition + + val podcast = playbackStateRelay.blockingFirst().podcast + // When the "skip last" option is enabled, we need to pause the chapter at the time configured in "skip last." + // Otherwise, this won't work with the sleep timer, as the sleep timer stops only when the chapter finishes + if (podcast != null && podcast.skipLastSecs > 0) { + pauseSuspend( + transientLoss = false, + sourceView = SourceView.AUTO_PAUSE, + transitionVersion = transitionVersion, + ) + if (!playerTransitions.isCurrent(transitionVersion)) return@transition + } + onPlayerPaused() - val podcast = playbackStateRelay.blockingFirst().podcast - // When the "skip last" option is enabled, we need to pause the chapter at the time configured in "skip last." - // Otherwise, this won't work with the sleep timer, as the sleep timer stops only when the chapter finishes - if (podcast != null && podcast.skipLastSecs > 0) { - pause(sourceView = SourceView.AUTO_PAUSE) + stop(transitionVersion) } - onPlayerPaused() - - stop() }, ) } @@ -2779,20 +3412,35 @@ open class PlaybackManager @Inject constructor( val episode = upNextQueue.currentEpisode ?: return val currentPlayer = this.player if (currentPlayer == null || episode.uuid != currentPlayer.episodeUuid) { - loadCurrentEpisode(false) + val transitionVersion = playerTransitions.beginTransition() + withTransitionCompletion(transitionVersion) { + loadCurrentEpisode(play = false, transitionVersion = transitionVersion) + } } } fun loadQueue(): Job { return launch { - val episode = upNextQueue.currentEpisode ?: return@launch - val currentPlayer = this@PlaybackManager.player - if (currentPlayer == null) { - withContext(Dispatchers.Main) { - updatePausedPlaybackState() + val episode = upNextQueue.currentEpisode + val currentPlayer = player + val shouldLoadEpisode = episode != null && + currentPlayer != null && + episode.uuid != currentPlayer.episodeUuid + val transitionVersion = if (shouldLoadEpisode) { + playerTransitions.beginTransition() + } else { + null + } + withOptionalTransitionCompletion(transitionVersion) { + if (episode == null) return@withOptionalTransitionCompletion + + if (currentPlayer == null) { + withContext(Dispatchers.Main) { + updatePausedPlaybackState() + } + } else if (transitionVersion != null) { + loadCurrentEpisode(play = false, transitionVersion = transitionVersion) } - } else if (episode.uuid != currentPlayer.episodeUuid) { - loadCurrentEpisode(false) } } } diff --git a/modules/services/repositories/src/main/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlayerTransitionCoordinator.kt b/modules/services/repositories/src/main/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlayerTransitionCoordinator.kt new file mode 100644 index 00000000000..ef73f09bbb1 --- /dev/null +++ b/modules/services/repositories/src/main/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlayerTransitionCoordinator.kt @@ -0,0 +1,274 @@ +package au.com.shiftyjelly.pocketcasts.repositories.playback + +import java.util.concurrent.atomic.AtomicLong +import java.util.concurrent.atomic.AtomicReference +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** + * Orders asynchronous player transitions and serializes access to the active [Player]. + * + * A transition receives its version when the command is issued. Work that finishes after a newer command is ignored, + * while current work runs exclusively so a player cannot be replaced while it is being configured or started. + */ +internal class PlayerTransitionCoordinator { + class Token internal constructor(internal val version: Long) + class Snapshot internal constructor(internal val version: Long) + class EventTransition internal constructor( + val token: Token, + internal val eventSourceToken: EventSourceToken, + ) + class EventSourceToken internal constructor( + internal val source: Any, + internal val generation: Long, + internal val transitionVersion: Long, + internal val isAcceptingEvents: Boolean, + ) + + private val mutex = Mutex() + private val stateLock = Any() + private val version = AtomicLong() + private val settledVersion = AtomicLong() + private val eventSourceGeneration = AtomicLong() + private val eventSource = AtomicReference() + private val stateRevision = MutableStateFlow(0L) + + fun beginTransition(): Token = Token(version.incrementAndGet()) + + fun snapshot(): Snapshot = Snapshot(version.get()) + + /** + * Starts a transition only when no newer transition has begun since [snapshot] was captured. + */ + fun tryBeginTransition(snapshot: Snapshot): Token? { + val nextVersion = snapshot.version + 1 + return if (version.compareAndSet(snapshot.version, nextVersion)) { + Token(nextVersion) + } else { + null + } + } + + fun isCurrent(token: Token): Boolean = token.version == version.get() + + /** + * Waits until the newest transition has completed and returns a snapshot that can be claimed without overtaking + * an in-flight command. + */ + suspend fun awaitSettledSnapshot(): Snapshot { + while (true) { + val observedRevision = stateRevision.value + val currentVersion = version.get() + if (settledVersion.get() == currentVersion) { + return Snapshot(currentVersion) + } + stateRevision.first { it != observedRevision } + } + } + + /** + * Starts a new generation for callbacks from [source]. + * + * Event-source generations change only when a player is created or configured, not when transition preparation + * begins. This keeps live-player events flowing during preparation while rejecting callbacks queued by an older + * configuration of the same player. + */ + fun bindEventSource(source: Any, token: Token) { + synchronized(stateLock) { + eventSource.set( + EventSourceToken( + source = source, + generation = eventSourceGeneration.incrementAndGet(), + transitionVersion = token.version, + isAcceptingEvents = true, + ), + ) + signalStateChange() + } + } + + fun tokenForEventSource(source: Any): EventSourceToken? { + return eventSource.get()?.takeIf { it.source === source && it.isAcceptingEvents } + } + + fun hasEventSource(source: Any): Boolean { + return eventSource.get()?.source === source + } + + /** + * Returns whether [source] is still reserved by a terminal event that predates [completingToken]. + * + * A newer no-op command must preserve that reservation while the terminal handler retries its final player + * transition. The terminal transition that owns the reservation must still validate the final queue/player state. + */ + fun hasInactiveEventSourceFromEarlierTransition( + source: Any, + completingToken: Token, + ): Boolean { + val current = eventSource.get() + return current?.source === source && + !current.isAcceptingEvents && + current.transitionVersion != completingToken.version + } + + fun isCurrentEventSource(token: EventSourceToken): Boolean { + val current = eventSource.get() + return current?.source === token.source && + current.generation == token.generation && + current.isAcceptingEvents + } + + fun clearEventSource(source: Any) { + synchronized(stateLock) { + val previous = eventSource.getAndUpdate { current -> + current?.takeUnless { it.source === source } + } + if (previous?.source === source) { + signalStateChange() + } + } + } + + /** + * Completes [token] after all earlier player commits have left the serialized section. + * + * An unchanged live source adopts the completed transition without changing its generation, so passive callbacks + * remain valid. A cleared or replaced source is never rebound by completion. + */ + suspend fun completeTransition( + token: Token, + sourceProvider: suspend () -> Any?, + ): Boolean = mutex.withLock { + if (!isCurrent(token)) { + false + } else { + val source = sourceProvider() + if (!isCurrent(token)) { + false + } else { + synchronized(stateLock) { + val current = eventSource.get() + if (source != null && current?.source === source) { + val shouldRemainInactive = !current.isAcceptingEvents && + current.transitionVersion != token.version + eventSource.set( + EventSourceToken( + source = source, + generation = current.generation, + transitionVersion = token.version, + isAcceptingEvents = !shouldRemainInactive, + ), + ) + } else { + eventSource.set(null) + } + settledVersion.set(token.version) + signalStateChange() + } + true + } + } + } + + /** + * Waits for in-flight commands that have not changed [eventSourceToken]'s source generation, then atomically + * claims the source for a terminal event. Claiming starts a new generation so callbacks queued beside the + * terminal event cannot update the next playback state. + */ + suspend fun beginTransitionForEventSource(eventSourceToken: EventSourceToken): EventTransition? { + return beginEventTransition(eventSourceToken, expectedToAcceptEvents = true) + } + + suspend fun retryTransitionForEventSource(eventSourceToken: EventSourceToken): EventTransition? { + return beginEventTransition(eventSourceToken, expectedToAcceptEvents = false) + } + + private suspend fun beginEventTransition( + eventSourceToken: EventSourceToken, + expectedToAcceptEvents: Boolean, + ): EventTransition? { + while (true) { + val observedRevision = stateRevision.value + val result = mutex.withLock { + synchronized(stateLock) { + val current = eventSource.get() + val isCurrentSource = current?.source === eventSourceToken.source && + current.generation == eventSourceToken.generation && + current.isAcceptingEvents == expectedToAcceptEvents + if (!isCurrentSource) { + return@synchronized EventTransitionResult.Stale + } + + val currentVersion = version.get() + if (current.transitionVersion != currentVersion || settledVersion.get() != currentVersion) { + return@synchronized EventTransitionResult.Pending + } + + val nextVersion = currentVersion + 1 + if (!version.compareAndSet(currentVersion, nextVersion)) { + return@synchronized EventTransitionResult.Pending + } + + val eventGeneration = eventSourceGeneration.incrementAndGet() + val claimedSource = EventSourceToken( + source = eventSourceToken.source, + generation = eventGeneration, + transitionVersion = nextVersion, + isAcceptingEvents = false, + ) + eventSource.set( + claimedSource, + ) + signalStateChange() + EventTransitionResult.Started( + EventTransition( + token = Token(nextVersion), + eventSourceToken = claimedSource, + ), + ) + } + } + when (result) { + EventTransitionResult.Pending -> stateRevision.first { it != observedRevision } + EventTransitionResult.Stale -> return null + is EventTransitionResult.Started -> return result.transition + } + } + } + + suspend fun runIfEventSourceCurrent( + token: EventSourceToken, + block: suspend () -> Unit, + ): Boolean = mutex.withLock { + if (!isCurrentEventSource(token)) { + false + } else { + block() + true + } + } + + suspend fun runIfCurrent( + token: Token, + block: suspend () -> Unit, + ): Boolean = mutex.withLock { + if (!isCurrent(token)) { + false + } else { + block() + true + } + } + + private fun signalStateChange() { + stateRevision.value = stateRevision.value + 1 + } + + private sealed interface EventTransitionResult { + data object Pending : EventTransitionResult + data object Stale : EventTransitionResult + data class Started(val transition: EventTransition) : EventTransitionResult + } +} diff --git a/modules/services/repositories/src/test/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlaybackManagerTransitionTest.kt b/modules/services/repositories/src/test/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlaybackManagerTransitionTest.kt new file mode 100644 index 00000000000..817fb27b1bc --- /dev/null +++ b/modules/services/repositories/src/test/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlaybackManagerTransitionTest.kt @@ -0,0 +1,490 @@ +@file:OptIn(ExperimentalCoroutinesApi::class) + +package au.com.shiftyjelly.pocketcasts.repositories.playback + +import androidx.work.testing.WorkManagerTestInitHelper +import au.com.shiftyjelly.pocketcasts.analytics.SourceView +import au.com.shiftyjelly.pocketcasts.models.entity.BaseEpisode +import au.com.shiftyjelly.pocketcasts.models.entity.PodcastEpisode +import au.com.shiftyjelly.pocketcasts.models.entity.UserEpisode +import au.com.shiftyjelly.pocketcasts.models.to.PlaybackEffects +import au.com.shiftyjelly.pocketcasts.models.type.EpisodeDownloadStatus +import au.com.shiftyjelly.pocketcasts.models.type.UserEpisodeServerStatus +import au.com.shiftyjelly.pocketcasts.preferences.Settings +import au.com.shiftyjelly.pocketcasts.preferences.UserSetting +import au.com.shiftyjelly.pocketcasts.repositories.bookmark.BookmarkManager +import au.com.shiftyjelly.pocketcasts.repositories.chromecast.CastManager +import au.com.shiftyjelly.pocketcasts.repositories.download.DownloadQueue +import au.com.shiftyjelly.pocketcasts.repositories.history.upnext.UpNextHistoryManager +import au.com.shiftyjelly.pocketcasts.repositories.notification.NotificationHelper +import au.com.shiftyjelly.pocketcasts.repositories.notification.NotificationManager +import au.com.shiftyjelly.pocketcasts.repositories.playlist.PlaylistManager +import au.com.shiftyjelly.pocketcasts.repositories.podcast.AlternateEnclosureManager +import au.com.shiftyjelly.pocketcasts.repositories.podcast.ChapterManager +import au.com.shiftyjelly.pocketcasts.repositories.podcast.EpisodeManager +import au.com.shiftyjelly.pocketcasts.repositories.podcast.PodcastManager +import au.com.shiftyjelly.pocketcasts.repositories.podcast.UserEpisodeManager +import au.com.shiftyjelly.pocketcasts.repositories.shownotes.ShowNotesManager +import au.com.shiftyjelly.pocketcasts.repositories.sync.SyncManager +import au.com.shiftyjelly.pocketcasts.repositories.user.StatsManager +import au.com.shiftyjelly.pocketcasts.sharedtest.MainCoroutineRule +import com.automattic.android.tracks.crashlogging.CrashLogging +import com.automattic.eventhorizon.EventHorizon +import io.reactivex.Maybe +import io.reactivex.Single +import java.io.IOException +import java.util.Date +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.cancel +import kotlinx.coroutines.joinAll +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout +import org.junit.After +import org.junit.Assert.assertNull +import org.junit.Assert.assertSame +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.kotlin.any +import org.mockito.kotlin.anyOrNull +import org.mockito.kotlin.doSuspendableAnswer +import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.times +import org.mockito.kotlin.verify +import org.mockito.kotlin.verifyNoInteractions +import org.mockito.kotlin.whenever +import org.robolectric.RobolectricTestRunner +import org.robolectric.RuntimeEnvironment + +@RunWith(RobolectricTestRunner::class) +class PlaybackManagerTransitionTest { + @get:Rule + val coroutineRule = MainCoroutineRule() + + private val settings = mock() + private val podcastManager = mock() + private val episodeManager = mock() + private val statsManager = mock() + private val playerFactory = mock() + private val castManager = mock() + private val playlistManager = mock() + private val downloadQueue = mock() + private val upNextQueue = mock() + private val notificationHelper = mock() + private val userEpisodeManager = mock() + private val eventHorizon = mock() + private val syncManager = mock() + private val bookmarkManager = mock() + private val showNotesManager = mock() + private val chapterManager = mock() + private val sleepTimer = mock() + private val networkWatcherFactory = mock() + private val crashLogging = mock() + private val upNextHistoryManager = mock() + private val notificationManager = mock() + private val autoPlaySelector = mock() + private val browseTreeProvider = mock() + private val alternateEnclosureManager = mock() + + private lateinit var applicationScope: CoroutineScope + private lateinit var playbackManager: PlaybackManager + + @Before + fun setUp() { + val application = RuntimeEnvironment.getApplication() + WorkManagerTestInitHelper.initializeTestWorkManager(application) + applicationScope = CoroutineScope(SupervisorJob() + coroutineRule.testDispatcher) + runBlocking { + whenever(castManager.isConnected()).thenReturn(false) + } + playbackManager = PlaybackManager( + settings = settings, + podcastManager = podcastManager, + episodeManager = episodeManager, + statsManager = statsManager, + playerManager = playerFactory, + castManager = castManager, + application = application, + playlistManager = playlistManager, + downloadQueue = downloadQueue, + upNextQueue = upNextQueue, + notificationHelper = notificationHelper, + userEpisodeManager = userEpisodeManager, + eventHorizon = eventHorizon, + syncManager = syncManager, + bookmarkManager = bookmarkManager, + showNotesManager = showNotesManager, + chapterManager = chapterManager, + sleepTimer = sleepTimer, + playbackManagerNetworkWatcherFactory = networkWatcherFactory, + applicationScope = applicationScope, + crashLogging = crashLogging, + upNextHistoryManager = upNextHistoryManager, + notificationManager = notificationManager, + autoPlaySelector = autoPlaySelector, + browseTreeProvider = browseTreeProvider, + alternateEnclosureManager = alternateEnclosureManager, + ) + } + + @After + fun tearDown() { + applicationScope.cancel() + } + + @Test + fun `remove is not dropped when episode becomes current behind a newer command`() = runTest { + val blocker = episode("blocker") + val target = episode("target") + val other = episode("other") + var currentEpisode: BaseEpisode? = other + whenever(upNextQueue.currentEpisode).thenAnswer { currentEpisode } + + val blockerEntered = CompletableDeferred() + val releaseBlocker = CompletableDeferred() + val targetRemoved = CompletableDeferred() + whenever(upNextQueue.removeEpisode(any(), any())).doSuspendableAnswer { invocation -> + when ((invocation.arguments[0] as BaseEpisode).uuid) { + blocker.uuid -> { + blockerEntered.complete(Unit) + releaseBlocker.await() + } + + target.uuid -> targetRemoved.complete(Unit) + } + Unit + } + + val blockerJob = requireNotNull( + playbackManager.removeEpisodeAsync(blocker, SourceView.UNKNOWN, userInitiated = false), + ) + awaitRealTime(blockerEntered) + + val targetJob = requireNotNull( + playbackManager.removeEpisodeAsync(target, SourceView.UNKNOWN, userInitiated = false), + ) + val pauseJob = async(start = CoroutineStart.UNDISPATCHED) { + playbackManager.pauseSuspend(transientLoss = true) + } + currentEpisode = target + releaseBlocker.complete(Unit) + + awaitRealTime(targetRemoved, blockerJob, targetJob, pauseJob) + verify(upNextQueue, times(1)).removeEpisode(target, false) + verifyNoInteractions(playerFactory) + } + + @Test + fun `current episode removal remains durable when superseded while waiting`() = runTest { + val blocker = episode("blocker") + val target = episode("target") + val next = episode("next") + var currentEpisode: BaseEpisode? = target + whenever(upNextQueue.currentEpisode).thenAnswer { currentEpisode } + val player = mock() + whenever(player.episodeUuid).thenReturn(target.uuid) + playbackManager.player = player + playbackManager.playbackStateRelay.accept( + PlaybackState(state = PlaybackState.State.PLAYING), + ) + whenever(upNextQueue.size).thenReturn(2) + val playerStopped = CompletableDeferred() + whenever(player.stop()).doSuspendableAnswer { + playerStopped.complete(Unit) + Unit + } + + val blockerEntered = CompletableDeferred() + val releaseBlocker = CompletableDeferred() + val targetRemoved = CompletableDeferred() + whenever(upNextQueue.removeEpisode(any(), any())).doSuspendableAnswer { invocation -> + when ((invocation.arguments[0] as BaseEpisode).uuid) { + blocker.uuid -> { + blockerEntered.complete(Unit) + releaseBlocker.await() + } + + target.uuid -> { + currentEpisode = next + targetRemoved.complete(Unit) + } + } + Unit + } + + val blockerJob = requireNotNull( + playbackManager.removeEpisodeAsync(blocker, SourceView.UNKNOWN, userInitiated = false), + ) + awaitRealTime(blockerEntered) + + val targetJob = requireNotNull( + playbackManager.removeEpisodeAsync(target, SourceView.UNKNOWN, userInitiated = false), + ) + val pauseJob = async(start = CoroutineStart.UNDISPATCHED) { + playbackManager.pauseSuspend(transientLoss = true) + } + releaseBlocker.complete(Unit) + + awaitRealTime(targetRemoved, blockerJob, targetJob, pauseJob) + awaitRealTime(playerStopped) + verify(upNextQueue, times(1)).removeEpisode(target, false) + verify(player, times(2)).pause() + verify(player, times(1)).stop() + assertNull(playbackManager.player) + verifyNoInteractions(playerFactory) + } + + @Test + fun `failed episode switch does not retain outgoing player`() = runTest { + val outgoingEpisode = episode("outgoing") + val requestedEpisode = episode("requested") + var currentEpisode: BaseEpisode? = outgoingEpisode + whenever(upNextQueue.currentEpisode).thenAnswer { currentEpisode } + val outgoingPlayer = mock() + whenever(outgoingPlayer.episodeUuid).thenReturn(outgoingEpisode.uuid) + playbackManager.player = outgoingPlayer + whenever(upNextQueue.playNow(any(), anyOrNull(), any(), anyOrNull())).doSuspendableAnswer { + currentEpisode = requestedEpisode + Unit + } + val outgoingPlayerStopped = CompletableDeferred() + whenever(outgoingPlayer.stop()).doSuspendableAnswer { + outgoingPlayerStopped.complete(Unit) + Unit + } + + playbackManager.playNowSuspend(requestedEpisode) + + awaitRealTime(outgoingPlayerStopped) + verify(outgoingPlayer, times(1)).pause() + verify(outgoingPlayer, times(1)).stop() + assertNull(playbackManager.player) + verifyNoInteractions(playerFactory) + } + + @Test + fun `failed player configuration tears down partially bound player`() = runTest { + val requestedEpisode = episode("requested").apply { + downloadStatus = EpisodeDownloadStatus.Downloaded + downloadedFilePath = "/tmp/requested.mp3" + } + whenever(upNextQueue.currentEpisode).thenReturn(requestedEpisode) + whenever(episodeManager.findByUuid(requestedEpisode.uuid)).thenReturn(requestedEpisode) + + val partiallyConfiguredPlayer = mock() + whenever(partiallyConfiguredPlayer.episodeUuid).thenReturn(null) + whenever(playerFactory.createSimplePlayer(any())).thenReturn(partiallyConfiguredPlayer) + whenever(partiallyConfiguredPlayer.setEpisode(any(), any())) + .thenThrow(IllegalStateException("Player configuration failed")) + + val failure = runCatching { + playbackManager.playQueueSuspend() + }.exceptionOrNull() + + if (failure !is IllegalStateException) { + throw requireNotNull(failure) + } + verify(partiallyConfiguredPlayer, times(1)).stop() + assertNull(playbackManager.player) + } + + @Test + fun `removing current episode does not deadlock when next cloud file fails`() = runTest { + val outgoingEpisode = episode("outgoing") + val cloudEpisode = UserEpisode( + uuid = "cloud", + publishedDate = Date(), + serverStatus = UserEpisodeServerStatus.UPLOADED, + ) + var currentEpisode: BaseEpisode? = outgoingEpisode + whenever(upNextQueue.currentEpisode).thenAnswer { currentEpisode } + whenever(upNextQueue.size).thenReturn(2) + whenever(upNextQueue.removeEpisode(any(), any())).doSuspendableAnswer { invocation -> + currentEpisode = when ((invocation.arguments[0] as BaseEpisode).uuid) { + outgoingEpisode.uuid -> cloudEpisode + cloudEpisode.uuid -> null + else -> currentEpisode + } + Unit + } + whenever(userEpisodeManager.findEpisodeByUuidRxMaybe(cloudEpisode.uuid)) + .thenReturn(Maybe.just(cloudEpisode)) + whenever(userEpisodeManager.getPlaybackUrlRxSingle(cloudEpisode)) + .thenReturn(Single.error(IOException("expired cloud URL"))) + val autoPlaySetting = mock>() + whenever(autoPlaySetting.value).thenReturn(false) + whenever(settings.autoPlayNextEpisodeOnEmpty).thenReturn(autoPlaySetting) + + val outgoingPlayer = mock() + whenever(outgoingPlayer.episodeUuid).thenReturn(outgoingEpisode.uuid) + playbackManager.player = outgoingPlayer + + val removalJob = requireNotNull( + playbackManager.removeEpisodeAsync(outgoingEpisode, SourceView.UNKNOWN, userInitiated = false), + ) + + awaitRealTime(removalJob) + + verify(upNextQueue, times(1)).removeEpisode(outgoingEpisode, false) + verify(upNextQueue, times(1)).removeEpisode(cloudEpisode, false) + assertNull(currentEpisode) + } + + @Test + fun `completion resumes loading next episode after newer no-op transition`() = runTest { + assertCompletionResumesAfterNoOp(CompletionTrigger.PlayerEvent) + } + + @Test + fun `skip past end resumes loading next episode after newer no-op transition`() = runTest { + assertCompletionResumesAfterNoOp(CompletionTrigger.SkipPastEnd) + } + + private suspend fun assertCompletionResumesAfterNoOp(trigger: CompletionTrigger) { + val completedEpisode = UserEpisode( + uuid = "completed", + publishedDate = Date(), + serverStatus = UserEpisodeServerStatus.UPLOADED, + ) + val nextEpisode = episode("next").apply { + downloadStatus = EpisodeDownloadStatus.Downloaded + downloadedFilePath = "/tmp/next.mp3" + } + var currentEpisode: BaseEpisode? = completedEpisode + whenever(upNextQueue.currentEpisode).thenAnswer { currentEpisode } + whenever(upNextQueue.removeEpisode(any(), any())).doSuspendableAnswer { + currentEpisode = nextEpisode + Unit + } + whenever(episodeManager.findByUuid(nextEpisode.uuid)).thenReturn(nextEpisode) + whenever(sleepTimer.state).thenReturn(SleepTimerState()) + + val disabledSetting = mock>() + whenever(disabledSetting.value).thenReturn(false) + whenever(settings.upNextShuffle).thenReturn(disabledSetting) + whenever(settings.audioOnly).thenReturn(disabledSetting) + val playbackEffectsSetting = mock>() + whenever(playbackEffectsSetting.value).thenReturn(PlaybackEffects()) + whenever(settings.globalPlaybackEffects).thenReturn(playbackEffectsSetting) + + var playerEpisodeUuid: String? = completedEpisode.uuid + val nextEpisodePaused = CompletableDeferred() + val outgoingPlayer = mock() + whenever(outgoingPlayer.episodeUuid).thenAnswer { playerEpisodeUuid } + whenever(outgoingPlayer.isRemote).thenReturn(true) + whenever(outgoingPlayer.getCurrentPositionMs()).doSuspendableAnswer { + if (playerEpisodeUuid == nextEpisode.uuid) { + nextEpisodePaused.complete(Unit) + } + 0 + } + playbackManager.player = outgoingPlayer + playbackManager.pauseSuspend(transientLoss = true) + playbackManager.playbackStateRelay.accept( + PlaybackState( + state = PlaybackState.State.PAUSED, + episodeUuid = completedEpisode.uuid, + ), + ) + + val completionPausedAfterRemoval = CompletableDeferred() + val releaseCompletion = CompletableDeferred() + whenever(userEpisodeManager.deletePlayedEpisodeIfReq(any(), any())).doSuspendableAnswer { + completionPausedAfterRemoval.complete(Unit) + releaseCompletion.await() + } + val nextEpisodeConfigured = CompletableDeferred() + whenever(outgoingPlayer.setEpisode(any(), any())).thenAnswer { invocation -> + playerEpisodeUuid = (invocation.arguments[0] as BaseEpisode).uuid + nextEpisodeConfigured.complete(Unit) + Unit + } + + when (trigger) { + CompletionTrigger.PlayerEvent -> { + dispatchPlayerEvent(outgoingPlayer, PlayerEvent.Completion(completedEpisode.uuid)) + } + + CompletionTrigger.SkipPastEnd -> { + whenever(outgoingPlayer.durationMs()).thenReturn(1_000) + playbackManager.skipForwardSuspend(jumpAmountSeconds = 30) + } + } + awaitRealTime(completionPausedAfterRemoval) + + playbackManager.skipForwardSuspend(jumpAmountSeconds = 30) + verify(outgoingPlayer, never()).stop() + + releaseCompletion.complete(Unit) + awaitRealTime(nextEpisodeConfigured) + awaitRealTime(nextEpisodePaused) + awaitPlaybackTransitionsSettled() + + verify(upNextQueue, times(1)).removeEpisode(completedEpisode, false) + verify(episodeManager, times(1)) + .updatePlayingStatusBlocking(completedEpisode, au.com.shiftyjelly.pocketcasts.models.type.EpisodePlayingStatus.COMPLETED) + verify(userEpisodeManager, times(1)).deletePlayedEpisodeIfReq(completedEpisode, playbackManager) + verify(outgoingPlayer, times(1)).setEpisode(nextEpisode, false) + assertSame(outgoingPlayer, playbackManager.player) + } + + private enum class CompletionTrigger { + PlayerEvent, + SkipPastEnd, + } + + private fun episode(uuid: String) = PodcastEpisode( + uuid = uuid, + publishedDate = Date(), + downloadUrl = "https://example.com/$uuid.mp3", + ) + + private fun dispatchPlayerEvent(player: Player, event: PlayerEvent) { + PlaybackManager::class.java + .getDeclaredMethod("onPlayerEvent", Player::class.java, PlayerEvent::class.java) + .apply { isAccessible = true } + .invoke(playbackManager, player, event) + } + + private suspend fun awaitPlaybackTransitionsSettled() { + val coordinator = PlaybackManager::class.java + .getDeclaredField("playerTransitions") + .apply { isAccessible = true } + .get(playbackManager) as PlayerTransitionCoordinator + withContext(Dispatchers.Default) { + withTimeout(5_000) { + coordinator.awaitSettledSnapshot() + } + } + } + + private suspend fun awaitRealTime( + deferred: CompletableDeferred, + vararg jobs: kotlinx.coroutines.Job, + ) { + withContext(Dispatchers.Default) { + withTimeout(5_000) { + deferred.await() + jobs.asList().joinAll() + } + } + } + + private suspend fun awaitRealTime(vararg jobs: kotlinx.coroutines.Job) { + withContext(Dispatchers.Default) { + withTimeout(5_000) { + jobs.asList().joinAll() + } + } + } +} diff --git a/modules/services/repositories/src/test/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlayerTransitionCoordinatorTest.kt b/modules/services/repositories/src/test/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlayerTransitionCoordinatorTest.kt new file mode 100644 index 00000000000..731e6eae785 --- /dev/null +++ b/modules/services/repositories/src/test/java/au/com/shiftyjelly/pocketcasts/repositories/playback/PlayerTransitionCoordinatorTest.kt @@ -0,0 +1,413 @@ +@file:OptIn(ExperimentalCoroutinesApi::class) + +package au.com.shiftyjelly.pocketcasts.repositories.playback + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.joinAll +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue +import org.junit.Test + +class PlayerTransitionCoordinatorTest { + private val coordinator = PlayerTransitionCoordinator() + + @Test + fun `newer transition wins when older preparation finishes last`() = runTest { + val olderPrepared = CompletableDeferred() + val releaseOlder = CompletableDeferred() + val commits = mutableListOf() + + val olderVersion = coordinator.beginTransition() + val older = async { + olderPrepared.complete(Unit) + releaseOlder.await() + coordinator.runIfCurrent(olderVersion) { + commits += "older" + } + } + olderPrepared.await() + + val newerVersion = coordinator.beginTransition() + val newerCommitted = coordinator.runIfCurrent(newerVersion) { + commits += "newer" + } + + releaseOlder.complete(Unit) + + assertTrue(newerCommitted) + assertFalse(older.await()) + assertEquals(listOf("newer"), commits) + } + + @Test + fun `commits for one transition are serialized`() = runTest { + val transitionVersion = coordinator.beginTransition() + val firstEntered = CompletableDeferred() + val releaseFirst = CompletableDeferred() + val commits = mutableListOf() + var activeCommits = 0 + var maxActiveCommits = 0 + + val first = launch { + coordinator.runIfCurrent(transitionVersion) { + activeCommits++ + maxActiveCommits = maxOf(maxActiveCommits, activeCommits) + firstEntered.complete(Unit) + releaseFirst.await() + commits += "first" + activeCommits-- + } + } + firstEntered.await() + + val second = launch { + coordinator.runIfCurrent(transitionVersion) { + activeCommits++ + maxActiveCommits = maxOf(maxActiveCommits, activeCommits) + commits += "second" + activeCommits-- + } + } + runCurrent() + + assertFalse(second.isCompleted) + + releaseFirst.complete(Unit) + joinAll(first, second) + + assertEquals(1, maxActiveCommits) + assertEquals(listOf("first", "second"), commits) + } + + @Test + fun `queued commit rechecks version after acquiring mutex`() = runTest { + val firstVersion = coordinator.beginTransition() + val firstEntered = CompletableDeferred() + val releaseFirst = CompletableDeferred() + val commits = mutableListOf() + + val first = async { + coordinator.runIfCurrent(firstVersion) { + firstEntered.complete(Unit) + releaseFirst.await() + commits += "first" + } + } + firstEntered.await() + + val queuedVersion = coordinator.beginTransition() + val queued = async { + coordinator.runIfCurrent(queuedVersion) { + commits += "queued" + } + } + runCurrent() + + val latestVersion = coordinator.beginTransition() + releaseFirst.complete(Unit) + + assertTrue(first.await()) + assertFalse(queued.await()) + assertTrue( + coordinator.runIfCurrent(latestVersion) { + commits += "latest" + }, + ) + assertEquals(listOf("first", "latest"), commits) + } + + @Test + fun `unchanged snapshot can begin a transition`() { + val snapshot = coordinator.snapshot() + + val transition = requireNotNull(coordinator.tryBeginTransition(snapshot)) + + assertTrue(coordinator.isCurrent(transition)) + } + + @Test + fun `snapshot cannot supersede a newer transition`() { + val snapshot = coordinator.snapshot() + val newerTransition = coordinator.beginTransition() + + assertNull(coordinator.tryBeginTransition(snapshot)) + assertTrue(coordinator.isCurrent(newerTransition)) + } + + @Test + fun `snapshot can begin only one transition`() { + val snapshot = coordinator.snapshot() + + val firstTransition = requireNotNull(coordinator.tryBeginTransition(snapshot)) + + assertNull(coordinator.tryBeginTransition(snapshot)) + assertTrue(coordinator.isCurrent(firstTransition)) + } + + @Test + fun `in-flight command does not invalidate callbacks from live player configuration`() = runTest { + val player = Any() + bindSettledSource(player) + val eventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + + coordinator.beginTransition() + + assertTrue(coordinator.isCurrentEventSource(eventSourceToken)) + } + + @Test + fun `terminal event waits for in-flight command and resumes after same-source completion`() = runTest { + val player = Any() + bindSettledSource(player) + val eventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + + val inFlightTransition = coordinator.beginTransition() + val terminalTransition = async { + coordinator.beginTransitionForEventSource(eventSourceToken) + } + runCurrent() + + assertFalse(terminalTransition.isCompleted) + assertTrue(coordinator.isCurrent(inFlightTransition)) + + assertTrue(coordinator.completeTransition(inFlightTransition) { player }) + + assertTrue(coordinator.isCurrent(requireNotNull(terminalTransition.await()).token)) + } + + @Test + fun `source reconfiguration invalidates terminal event waiting behind a command`() = runTest { + val player = Any() + bindSettledSource(player) + val oldEventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + + val reconfiguration = coordinator.beginTransition() + val terminalTransition = async { + coordinator.beginTransitionForEventSource(oldEventSourceToken) + } + runCurrent() + + assertTrue( + coordinator.runIfCurrent(reconfiguration) { + coordinator.bindEventSource(player, reconfiguration) + }, + ) + assertTrue(coordinator.completeTransition(reconfiguration) { player }) + + assertNull(terminalTransition.await()) + } + + @Test + fun `same source completion preserves passive events and allows waiting terminal event`() = runTest { + val player = Any() + bindSettledSource(player) + val oldEventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + + val completedTransition = coordinator.beginTransition() + assertTrue(coordinator.completeTransition(completedTransition) { player }) + + assertTrue(coordinator.isCurrentEventSource(oldEventSourceToken)) + assertTrue( + coordinator.isCurrent( + requireNotNull(coordinator.beginTransitionForEventSource(oldEventSourceToken)).token, + ), + ) + } + + @Test + fun `completion waits for in-progress player commit before adopting source`() = runTest { + val player = Any() + val playerCommitEntered = CompletableDeferred() + val releasePlayerCommit = CompletableDeferred() + val playerTransition = coordinator.beginTransition() + val playerCommit = async { + coordinator.runIfCurrent(playerTransition) { + playerCommitEntered.complete(Unit) + releasePlayerCommit.await() + coordinator.bindEventSource(player, playerTransition) + } + } + playerCommitEntered.await() + + val completedTransition = coordinator.beginTransition() + val completion = async { + coordinator.completeTransition(completedTransition) { player } + } + runCurrent() + + assertFalse(completion.isCompleted) + + releasePlayerCommit.complete(Unit) + + assertTrue(playerCommit.await()) + assertTrue(completion.await()) + assertEquals( + completedTransition.version, + requireNotNull(coordinator.tokenForEventSource(player)).transitionVersion, + ) + } + + @Test + fun `stale completion cannot overwrite newer transition state`() = runTest { + val player = Any() + val committedTransition = coordinator.beginTransition() + coordinator.bindEventSource(player, committedTransition) + val committedEventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + + val staleTransition = coordinator.beginTransition() + val currentTransition = coordinator.beginTransition() + assertFalse(coordinator.completeTransition(staleTransition) { player }) + + assertSame(committedEventSourceToken, coordinator.tokenForEventSource(player)) + assertTrue(coordinator.isCurrent(currentTransition)) + } + + @Test + fun `completion does not adopt an unbound replacement source`() = runTest { + val previousPlayer = Any() + val replacementPlayer = Any() + val previousTransition = coordinator.beginTransition() + coordinator.bindEventSource(previousPlayer, previousTransition) + + val replacementTransition = coordinator.beginTransition() + + assertTrue(coordinator.completeTransition(replacementTransition) { replacementPlayer }) + assertNull(coordinator.tokenForEventSource(previousPlayer)) + assertNull(coordinator.tokenForEventSource(replacementPlayer)) + } + + @Test + fun `duplicate terminal event can begin only one transition`() = runTest { + val player = Any() + bindSettledSource(player) + val eventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + + val firstTerminalTransition = requireNotNull(coordinator.beginTransitionForEventSource(eventSourceToken)) + + assertNull(coordinator.beginTransitionForEventSource(eventSourceToken)) + assertNull(coordinator.tokenForEventSource(player)) + assertTrue(coordinator.isCurrent(firstTerminalTransition.token)) + } + + @Test + fun `terminal source accepts callbacks again only after transition completion`() = runTest { + val player = Any() + bindSettledSource(player) + val eventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + + val terminalTransition = requireNotNull(coordinator.beginTransitionForEventSource(eventSourceToken)) + + assertNull(coordinator.tokenForEventSource(player)) + assertFalse( + coordinator.runIfEventSourceCurrent(eventSourceToken) { + error("Outgoing source callback should not run during a terminal transition") + }, + ) + + assertTrue(coordinator.completeTransition(terminalTransition.token) { player }) + assertTrue( + coordinator.isCurrentEventSource( + requireNotNull(coordinator.tokenForEventSource(player)), + ), + ) + } + + @Test + fun `terminal claim survives a newer same-source no-op transition`() = runTest { + val player = Any() + bindSettledSource(player) + val eventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + val terminalTransition = requireNotNull(coordinator.beginTransitionForEventSource(eventSourceToken)) + + val noOpTransition = coordinator.beginTransition() + assertTrue(coordinator.hasInactiveEventSourceFromEarlierTransition(player, noOpTransition)) + assertFalse(coordinator.hasInactiveEventSourceFromEarlierTransition(player, terminalTransition.token)) + assertTrue(coordinator.completeTransition(noOpTransition) { player }) + + assertNull(coordinator.tokenForEventSource(player)) + val retriedTransition = requireNotNull( + coordinator.retryTransitionForEventSource(terminalTransition.eventSourceToken), + ) + assertTrue(coordinator.isCurrent(retriedTransition.token)) + + assertTrue(coordinator.completeTransition(retriedTransition.token) { player }) + requireNotNull(coordinator.tokenForEventSource(player)) + } + + @Test + fun `passive event finishes before source can be rebound`() = runTest { + val player = Any() + bindSettledSource(player) + val eventSourceToken = requireNotNull(coordinator.tokenForEventSource(player)) + val passiveEventEntered = CompletableDeferred() + val releasePassiveEvent = CompletableDeferred() + val passiveEvent = async { + coordinator.runIfEventSourceCurrent(eventSourceToken) { + passiveEventEntered.complete(Unit) + releasePassiveEvent.await() + } + } + passiveEventEntered.await() + + val reconfiguration = coordinator.beginTransition() + val rebind = async { + coordinator.runIfCurrent(reconfiguration) { + coordinator.bindEventSource(player, reconfiguration) + } + } + runCurrent() + + assertFalse(rebind.isCompleted) + + releasePassiveEvent.complete(Unit) + + assertTrue(passiveEvent.await()) + assertTrue(rebind.await()) + assertFalse(coordinator.isCurrentEventSource(eventSourceToken)) + } + + @Test + fun `reconfiguring same player invalidates callbacks from previous generation`() { + val player = Any() + val previousTransition = coordinator.beginTransition() + coordinator.bindEventSource(player, previousTransition) + val previousGeneration = requireNotNull(coordinator.tokenForEventSource(player)) + + val currentTransition = coordinator.beginTransition() + coordinator.bindEventSource(player, currentTransition) + val currentGeneration = requireNotNull(coordinator.tokenForEventSource(player)) + + assertFalse(coordinator.isCurrentEventSource(previousGeneration)) + assertTrue(coordinator.isCurrentEventSource(currentGeneration)) + } + + @Test + fun `replacing player invalidates callbacks from previous source`() { + val previousPlayer = Any() + val currentPlayer = Any() + val previousTransition = coordinator.beginTransition() + coordinator.bindEventSource(previousPlayer, previousTransition) + val previousSource = requireNotNull(coordinator.tokenForEventSource(previousPlayer)) + + val currentTransition = coordinator.beginTransition() + coordinator.bindEventSource(currentPlayer, currentTransition) + + assertFalse(coordinator.isCurrentEventSource(previousSource)) + assertNull(coordinator.tokenForEventSource(previousPlayer)) + assertTrue(coordinator.isCurrentEventSource(requireNotNull(coordinator.tokenForEventSource(currentPlayer)))) + } + + private suspend fun bindSettledSource(source: Any) { + val transition = coordinator.beginTransition() + coordinator.bindEventSource(source, transition) + assertTrue(coordinator.completeTransition(transition) { source }) + } +}