From cf74736579f44c7d9e8715b6189454e503757ee9 Mon Sep 17 00:00:00 2001 From: Damontecres Date: Tue, 29 Sep 2026 12:43:13 -0400 Subject: [PATCH] Listen for user config changes and improved websocket error handling (#2012) ## Description Add better error handling for server web socket messages that are (technically) invalid. For example, I talked to Niels and some third party plugins don't always have the required fields. Since the server isn't as strict as the client SDK, it's fine on the server but will generate an invalid message sent to the client. Also listens for server-side user configuration changes via web socket. ### Related issues Fixes #1979 ### Testing Valid server messages are handled as normal and I did some manual testing to push invalid JSON into the webSocketApi, but the code is gnarly and not viable as an actual unit test ## Screenshots N/A ## AI or LLM usage None --- .../wholphin/data/ServerRepository.kt | 12 +- .../wholphin/services/ServerEventListener.kt | 84 ++++++++--- .../wholphin/ui/playback/PlaybackViewModel.kt | 134 +++++++++++------- 3 files changed, 152 insertions(+), 78 deletions(-) diff --git a/app/src/main/java/com/github/damontecres/wholphin/data/ServerRepository.kt b/app/src/main/java/com/github/damontecres/wholphin/data/ServerRepository.kt index a86335be..f3cd888e 100644 --- a/app/src/main/java/com/github/damontecres/wholphin/data/ServerRepository.kt +++ b/app/src/main/java/com/github/damontecres/wholphin/data/ServerRepository.kt @@ -29,6 +29,7 @@ import org.jellyfin.sdk.api.client.extensions.quickConnectApi import org.jellyfin.sdk.api.client.extensions.systemApi import org.jellyfin.sdk.api.client.extensions.userApi import org.jellyfin.sdk.model.api.AuthenticationResult +import org.jellyfin.sdk.model.api.UserDto import org.jellyfin.sdk.model.serializer.toUUIDOrNull import timber.log.Timber import java.time.ZonedDateTime @@ -318,8 +319,17 @@ class ServerRepository */ suspend fun updateUserDto() { val userDto by apiClient.userApi.getCurrentUser() + updateUserDto(userDto) + } + + /** + * Update [currentUserDto] with the specified [UserDto] + * + * This will only update if the [UserDto] is for the [currentUser] + */ + fun updateUserDto(userDto: UserDto) { _currentUserDto.update { - if (it?.id == userDto.id && currentUser?.id == userDto.id) ServerUserConfig(userDto) else it + if (currentUser?.id == userDto.id) ServerUserConfig(userDto) else it } } diff --git a/app/src/main/java/com/github/damontecres/wholphin/services/ServerEventListener.kt b/app/src/main/java/com/github/damontecres/wholphin/services/ServerEventListener.kt index 708eb8b5..ca6cceb8 100644 --- a/app/src/main/java/com/github/damontecres/wholphin/services/ServerEventListener.kt +++ b/app/src/main/java/com/github/damontecres/wholphin/services/ServerEventListener.kt @@ -10,20 +10,25 @@ import com.github.damontecres.wholphin.data.ServerRepository import com.github.damontecres.wholphin.data.model.JellyfinServer import com.github.damontecres.wholphin.data.model.JellyfinUser import com.github.damontecres.wholphin.ui.collectLatestIn +import com.github.damontecres.wholphin.ui.launchDefault import com.github.damontecres.wholphin.ui.launchIO import com.github.damontecres.wholphin.ui.showToast import dagger.hilt.android.qualifiers.ActivityContext import dagger.hilt.android.scopes.ActivityScoped +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Job +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.isActive import org.jellyfin.sdk.api.client.ApiClient import org.jellyfin.sdk.api.client.extensions.sessionApi import org.jellyfin.sdk.api.sockets.subscribe import org.jellyfin.sdk.model.api.GeneralCommandMessage import org.jellyfin.sdk.model.api.GeneralCommandType import org.jellyfin.sdk.model.api.MediaType +import org.jellyfin.sdk.model.api.UserUpdatedMessage import timber.log.Timber import javax.inject.Inject @@ -68,36 +73,71 @@ class ServerEventListener ), supportsMediaControl = true, ) - setupListeners() + subscribeToWebSocket() } } } - fun setupListeners() { - serverRepository.currentUser + fun subscribeToWebSocket() { Timber.v("Subscribing to WebSocket") listenJob?.cancel() listenJob = - api.webSocket - .subscribe() - .onEach { message -> - if (message.data?.name in - setOf( - GeneralCommandType.DISPLAY_MESSAGE, - GeneralCommandType.SEND_STRING, - ) - ) { - val header = message.data?.arguments["Header"] - val text = - message.data?.arguments["Text"] ?: message.data?.arguments["String"] - val toast = - listOfNotNull(header, text) - .joinToString("\n") - showToast(context, toast, Toast.LENGTH_LONG) + activity.lifecycleScope.launchDefault { + try { + // Launch multiple listeners, but stop all if one fails + coroutineScope { + api.webSocket + .subscribe() + .onEach { message -> + Timber.v( + "Got GeneralCommandMessage: %s", + message.data?.name, + ) + when (message.data?.name) { + GeneralCommandType.DISPLAY_MESSAGE, + GeneralCommandType.SEND_STRING, + -> { + val header = message.data?.arguments["Header"] + val text = + message.data?.arguments["Text"] + ?: message.data?.arguments["String"] + val toast = + listOfNotNull(header, text) + .joinToString("\n") + if (toast.isNotBlank()) { + showToast(context, toast, Toast.LENGTH_LONG) + } + } + + else -> { + Timber.v( + "Ignoring GeneralCommandMessage: %s", + message.data?.name, + ) + } + } + }.catch { ex -> + Timber.e(ex, "Error in general message websocket subscription") + }.launchIn(this@coroutineScope) + + api.webSocket + .subscribe() + .catch { ex -> + Timber.e(ex, "Error in user updated websocket subscription") + }.collectLatestIn(this@coroutineScope) { msg -> + Timber.v("Got updated user: %s", msg.data?.id) + msg.data?.let { serverRepository.updateUserDto(it) } + } } - }.catch { ex -> - Timber.e(ex, "Error in websocket subscription") - }.launchIn(activity.lifecycleScope) + } catch (ex: CancellationException) { + throw ex + } catch (ex: Exception) { + Timber.e(ex, "Error in websocket connection") + if (activity.lifecycleScope.isActive) { + subscribeToWebSocket() + } + } + } } override fun onResume(owner: LifecycleOwner) { diff --git a/app/src/main/java/com/github/damontecres/wholphin/ui/playback/PlaybackViewModel.kt b/app/src/main/java/com/github/damontecres/wholphin/ui/playback/PlaybackViewModel.kt index 79d2ae57..012038a8 100644 --- a/app/src/main/java/com/github/damontecres/wholphin/ui/playback/PlaybackViewModel.kt +++ b/app/src/main/java/com/github/damontecres/wholphin/ui/playback/PlaybackViewModel.kt @@ -18,6 +18,7 @@ import androidx.media3.common.Player import androidx.media3.common.Tracks import androidx.media3.common.text.CueGroup import androidx.media3.common.util.UnstableApi +import androidx.media3.common.util.Util import androidx.media3.exoplayer.DecoderCounters import androidx.media3.exoplayer.DecoderReuseEvaluation import androidx.media3.exoplayer.ExoPlayer @@ -81,6 +82,7 @@ import dagger.hilt.android.lifecycle.HiltViewModel import dagger.hilt.android.qualifiers.ApplicationContext import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Job +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow @@ -113,6 +115,7 @@ import org.jellyfin.sdk.model.api.PlayMethod import org.jellyfin.sdk.model.api.PlaybackInfoDto import org.jellyfin.sdk.model.api.PlaystateCommand import org.jellyfin.sdk.model.api.PlaystateMessage +import org.jellyfin.sdk.model.api.PlaystateRequest import org.jellyfin.sdk.model.api.TrickplayInfo import org.jellyfin.sdk.model.api.VideoRange import org.jellyfin.sdk.model.api.VideoRangeType @@ -187,6 +190,7 @@ class PlaybackViewModel private var activityListener: TrackActivityPlaybackListener? = null private var trackChangeListener: TracksChangedListener? = null private val jobs = mutableListOf() + private var subscribeJob: Job? = null private val isPlaylist = destination is Destination.PlaybackList @@ -271,7 +275,7 @@ class PlaybackViewModel private fun configurePlayer() { player.addListener(this) (player as? ExoPlayer)?.addAnalyticsListener(this) - jobs.add(subscribe()) + subscribeToWebSocket() jobs.add(listenForTranscodeReason()) val sessionPlayer = MediaSessionPlayer( @@ -1417,63 +1421,83 @@ class PlaybackViewModel activityListener = null } - fun subscribe(): Job = - api.webSocket - .subscribe() - .onEach { message -> - message.data?.let { - withContext(WholphinDispatchers.Main) { - when (it.command) { - PlaystateCommand.STOP -> { - release() - navigationManager.goBack() - } - - PlaystateCommand.PAUSE -> { - player.pause() - } - - PlaystateCommand.UNPAUSE -> { - player.play() - } - - PlaystateCommand.NEXT_TRACK -> { - playNextUp() - } - - PlaystateCommand.PREVIOUS_TRACK -> { - playPrevious() - } - - PlaystateCommand.SEEK -> { - it.seekPositionTicks?.ticks?.let { - player.seekTo( - it.inWholeMilliseconds, - ) + fun subscribeToWebSocket() { + subscribeJob?.cancel() + try { + subscribeJob = + viewModelScope.launch { + coroutineScope { + api.webSocket + .subscribe() + .onEach { message -> + message.data?.let { request -> + Timber.v("Received playstate request: %s", request) + withContext(WholphinDispatchers.Main) { + handlePlaystateRequest(request) + } } - } - - PlaystateCommand.REWIND -> { - player.seekBack( - preferences.appPreferences.playbackPreferences.skipBackMs.milliseconds, - ) - } - - PlaystateCommand.FAST_FORWARD -> { - player.seekForward( - preferences.appPreferences.playbackPreferences.skipForwardMs.milliseconds, - ) - } - - PlaystateCommand.PLAY_PAUSE -> { - if (player.isPlaying) player.pause() else player.play() - } - } + }.catch { ex -> + Timber.e(ex, "Error in websocket subscription") + }.launchIn(this) } } - }.catch { ex -> - Timber.e(ex, "Error in websocket subscription") - }.launchIn(viewModelScope) + } catch (ex: CancellationException) { + throw ex + } catch (ex: Exception) { + Timber.e(ex, "Error in playback websocket subscription") + if (viewModelScope.isActive) { + subscribeToWebSocket() + } + } + } + + private suspend fun handlePlaystateRequest(request: PlaystateRequest) = + withContext(WholphinDispatchers.Main) { + when (request.command) { + PlaystateCommand.STOP -> { + release() + navigationManager.goBack() + } + + PlaystateCommand.PAUSE -> { + player.pause() + } + + PlaystateCommand.UNPAUSE -> { + player.play() + } + + PlaystateCommand.NEXT_TRACK -> { + playNextUp() + } + + PlaystateCommand.PREVIOUS_TRACK -> { + playPrevious() + } + + PlaystateCommand.SEEK -> { + request.seekPositionTicks?.ticks?.let { + player.seekTo(it.inWholeMilliseconds) + } + } + + PlaystateCommand.REWIND -> { + player.seekBack( + preferences.appPreferences.playbackPreferences.skipBackMs.milliseconds, + ) + } + + PlaystateCommand.FAST_FORWARD -> { + player.seekForward( + preferences.appPreferences.playbackPreferences.skipForwardMs.milliseconds, + ) + } + + PlaystateCommand.PLAY_PAUSE -> { + Util.handlePlayPauseButtonAction(player) + } + } + } /** * Atomically update [currentMediaInfo]