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

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
             }
         }
 

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() {
+        fun subscribeToWebSocket() {
-            serverRepository.currentUser
             Timber.v("Subscribing to WebSocket")
             listenJob?.cancel()
             listenJob =
-                api.webSocket
+                activity.lifecycleScope.launchDefault {
-                    .subscribe<GeneralCommandMessage>()
+                    try {
-                    .onEach { message ->
+                        // Launch multiple listeners, but stop all if one fails
-                        if (message.data?.name in
+                        coroutineScope {
-                            setOf(
+                            api.webSocket
-                                GeneralCommandType.DISPLAY_MESSAGE,
+                                .subscribe<GeneralCommandMessage>()
-                                GeneralCommandType.SEND_STRING,
+                                .onEach { message ->
-                            )
+                                    Timber.v(
-                        ) {
+                                        "Got GeneralCommandMessage: %s",
-                            val header = message.data?.arguments["Header"]
+                                        message.data?.name,
-                            val text =
+                                    )
-                                message.data?.arguments["Text"] ?: message.data?.arguments["String"]
+                                    when (message.data?.name) {
-                            val toast =
+                                        GeneralCommandType.DISPLAY_MESSAGE,
-                                listOfNotNull(header, text)
+                                        GeneralCommandType.SEND_STRING,
-                                    .joinToString("\n")
+                                        -> {
-                            showToast(context, toast, Toast.LENGTH_LONG)
+                                            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<UserUpdatedMessage>()
+                                .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 ->
+                    } catch (ex: CancellationException) {
-                        Timber.e(ex, "Error in websocket subscription")
+                        throw ex
-                    }.launchIn(activity.lifecycleScope)
+                    } catch (ex: Exception) {
+                        Timber.e(ex, "Error in websocket connection")
+                        if (activity.lifecycleScope.isActive) {
+                            subscribeToWebSocket()
+                        }
+                    }
+                }
         }
 
         override fun onResume(owner: LifecycleOwner) {

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<Job>()
+        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 =
+        fun subscribeToWebSocket() {
-            api.webSocket
+            subscribeJob?.cancel()
-                .subscribe<PlaystateMessage>()
+            try {
-                .onEach { message ->
+                subscribeJob =
-                    message.data?.let {
+                    viewModelScope.launch {
-                        withContext(WholphinDispatchers.Main) {
+                        coroutineScope {
-                            when (it.command) {
+                            api.webSocket
-                                PlaystateCommand.STOP -> {
+                                .subscribe<PlaystateMessage>()
-                                    release()
+                                .onEach { message ->
-                                    navigationManager.goBack()
+                                    message.data?.let { request ->
-                                }
+                                        Timber.v("Received playstate request: %s", request)
-
+                                        withContext(WholphinDispatchers.Main) {
-                                PlaystateCommand.PAUSE -> {
+                                            handlePlaystateRequest(request)
-                                    player.pause()
+                                        }
-                                }
-
-                                PlaystateCommand.UNPAUSE -> {
-                                    player.play()
-                                }
-
-                                PlaystateCommand.NEXT_TRACK -> {
-                                    playNextUp()
-                                }
-
-                                PlaystateCommand.PREVIOUS_TRACK -> {
-                                    playPrevious()
-                                }
-
-                                PlaystateCommand.SEEK -> {
-                                    it.seekPositionTicks?.ticks?.let {
-                                        player.seekTo(
-                                            it.inWholeMilliseconds,
-                                        )
                                     }
-                                }
+                                }.catch { ex ->
-
+                                    Timber.e(ex, "Error in websocket subscription")
-                                PlaystateCommand.REWIND -> {
+                                }.launchIn(this)
-                                    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 ->
+            } catch (ex: CancellationException) {
-                    Timber.e(ex, "Error in websocket subscription")
+                throw ex
-                }.launchIn(viewModelScope)
+            } 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]