Modify state management so that the app's screen can reconnect no matter what event causes it to regenerate.

Author: jaeyunn15

thunder-okhttp/src/main/java/com/jeremy/thunder/OkHttpWebSocket.kt

@@ -19,7 +19,7 @@ class OkHttpWebSocket internal constructor(
     private val scope: CoroutineScope
 ) : WebSocket {
 
-    private val _event = MutableStateFlow<com.jeremy.thunder.event.WebSocketEvent?>(null)
+    private val _event = MutableStateFlow<WebSocketEvent?>(null)
     override fun open() {
         socketListener.collectEvent().onStart {
             provider.provide(socketListener)
@@ -34,7 +34,7 @@ class OkHttpWebSocket internal constructor(
         }.launchIn(scope)
     }
 
-    override fun events(): Flow<com.jeremy.thunder.event.WebSocketEvent> {
+    override fun events(): Flow<WebSocketEvent> {
         return _event.asSharedFlow().filterNotNull()
     }
 

thunder/src/main/java/com/jeremy/thunder/connection/AppConnectionProvider.kt

@@ -4,7 +4,8 @@ import android.app.Activity
 import android.app.Application
 import android.os.Bundle
 import com.jeremy.thunder.state.ActivityState
-import com.jeremy.thunder.state.GetReady
+import com.jeremy.thunder.state.Background
+import com.jeremy.thunder.state.Foreground
 import com.jeremy.thunder.state.Initialize
 import com.jeremy.thunder.state.ShutDown
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -20,22 +21,24 @@ class AppConnectionProvider : AppConnectionListener, Application.ActivityLifecyc
     }
 
     override fun onActivityCreated(p0: Activity, p1: Bundle?) {
-        _eventFlow.tryEmit(GetReady)
+        _eventFlow.tryEmit(Initialize)
     }
 
     override fun onActivityStarted(p0: Activity) {
-        _eventFlow.tryEmit(GetReady)
+        _eventFlow.tryEmit(Foreground)
     }
 
     override fun onActivityResumed(p0: Activity) {
-        _eventFlow.tryEmit(GetReady)
+        _eventFlow.tryEmit(Foreground)
     }
 
     override fun onActivityPaused(p0: Activity) {}
 
     override fun onActivityStopped(p0: Activity) {}
 
-    override fun onActivitySaveInstanceState(p0: Activity, p1: Bundle) {}
+    override fun onActivitySaveInstanceState(p0: Activity, p1: Bundle) {
+        _eventFlow.tryEmit(Background)
+    }
 
     override fun onActivityDestroyed(p0: Activity) {
         _eventFlow.tryEmit(ShutDown)

thunder/src/main/java/com/jeremy/thunder/internal/ThunderStateManager.kt

@@ -7,13 +7,16 @@ import com.jeremy.thunder.connection.AppConnectionListener
 import com.jeremy.thunder.coroutine.CoroutineScope.scope
 import com.jeremy.thunder.event.WebSocketEvent
 import com.jeremy.thunder.network.NetworkConnectivityService
-import com.jeremy.thunder.state.GetReady
+import com.jeremy.thunder.state.Background
+import com.jeremy.thunder.state.Foreground
 import com.jeremy.thunder.state.Initialize
 import com.jeremy.thunder.state.NetworkState
 import com.jeremy.thunder.state.ShutDown
 import com.jeremy.thunder.state.ThunderError
 import com.jeremy.thunder.state.ThunderState
+import com.jeremy.thunder.thunderLog
 import com.jeremy.thunder.ws.WebSocket
+import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
@@ -22,9 +25,10 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.combine
-import kotlinx.coroutines.flow.getAndUpdate
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.update
+import kotlinx.coroutines.plus
 
 /**
  * Manage ThunderState as SocketState using NetworkState
@@ -39,19 +43,23 @@ class ThunderStateManager private constructor(
     private val webSocketCore: WebSocket.Factory,
     private val scope: CoroutineScope
 ) {
+    private val innerScope = scope + CoroutineExceptionHandler { _, throwable ->
+        thunderLog("[ThunderStateManager] = ${throwable.message}")
+    }
+
     private var socket: WebSocket? = null
 
     private val _socketState = MutableStateFlow<ThunderState>(ThunderState.IDLE)
 
     private val _events = MutableSharedFlow<WebSocketEvent>(replay = 1)
 
-    private var _lastSocketState: ThunderState = ThunderState.IDLE
-
     /**
      * If the device loses the network or the socket connection fails, it enters the error state below.
      * This is used to use the cache for recovery when a [ThunderState.CONNECTED] is reached.
      * */
-    private var isFromError = false
+    private var isReSubscription = false
+
+    private val _retryNeedFlag = MutableStateFlow<Boolean>(false)
 
     fun thunderStateAsFlow() = _socketState.asStateFlow()
 
@@ -61,85 +69,98 @@ class ThunderStateManager private constructor(
 
     init {
         /**
-         * The following code is used to open the valve based on the socket state.
-        * */
-        _socketState.onEach {
-            if (it is ThunderState.ERROR && networkState.hasAvailableNetworks()) {
-                closeConnection()
-                delay(500)
-                openConnection()
-            }
-            valveCache.onUpdateValveState(it)
-        }.launchIn(scope)
-
-        /**
-         * When an app is present in a process but offscreen, it automatically controls the connection based on two states to maintain the connection in the meantime.
+         * Update SocketState as ThunderState.
          * */
-        connectionListener.collectState().onEach {
-            when(it) {
-                Initialize -> Unit
-                GetReady -> openConnection()
-                ShutDown -> closeConnection()
+        _events.onEach { event ->
+            when (event) {
+                is WebSocketEvent.OnConnectionOpen -> {
+                    _socketState.update { ThunderState.CONNECTED }
+                }
+
+                is WebSocketEvent.OnMessageReceived -> Unit
+
+                WebSocketEvent.OnConnectionClosed -> {
+                    _socketState.update { ThunderState.DISCONNECTED }
+                }
+
+                is WebSocketEvent.OnConnectionError -> {
+                    isReSubscription = true
+                    _socketState.update { ThunderState.ERROR(ThunderError.SocketLoss(event.error)) }
+                }
             }
-        }.launchIn(scope)
+        }.launchIn(innerScope)
 
         /**
-         * Used to change the ThunderState based on the device's network connection status.
-        * */
-        networkState.networkStatus.onEach {
-            when (it) {
+         * Open, Retry Connection work as network state.
+         * */
+        combine(
+            _retryNeedFlag,
+            networkState.networkStatus
+        ) { retry, network ->
+            when (network) {
                 NetworkState.Available -> {
-                    openConnection()
+                    if (retry) {
+                        retryConnection()
+                    } else {
+                        openConnection()
+                    }
                 }
-
                 NetworkState.Unavailable -> {
-                    isFromError = true
-                    _socketState.updateThunderState(ThunderState.ERROR(ThunderError.NetworkLoss(null)))
-                    closeConnection()
+                    _socketState.update { ThunderState.ERROR(ThunderError.NetworkLoss) }
                 }
             }
-        }.launchIn(scope)
+        }.launchIn(innerScope)
 
-        _events.onEach {
-            when (it) {
-                is WebSocketEvent.OnConnectionOpen -> {
-                    _socketState.updateThunderState(ThunderState.CONNECTED)
+        /**
+         * Update RetryFlag And Valve And request socket message as upstream state.
+         * */
+        combine(
+            _socketState,
+            connectionListener.collectState()
+        ) { socketState, appState ->
+            when (appState) {
+                Initialize -> {
+                    openConnection()
                 }
-
-                is WebSocketEvent.OnMessageReceived -> Unit
-
-                WebSocketEvent.OnConnectionClosed -> {
-                    _socketState.updateThunderState(ThunderState.DISCONNECTED)
+                Foreground -> {
+                    if (socketState is ThunderState.ERROR) {
+                        _retryNeedFlag.update { true }
+                    }
                 }
-
-                is WebSocketEvent.OnConnectionError -> {
-                    isFromError = true
-                    _socketState.updateThunderState(ThunderState.ERROR(ThunderError.SocketLoss(it.error)))
+                Background -> {}
+                ShutDown -> {
+                    closeConnection()
                 }
             }
-        }.launchIn(scope)
+            valveCache.onUpdateValveState(socketState)
+        }.launchIn(innerScope)
 
         combine(
             _socketState,
             valveCache.emissionOfValveFlow()
-        ) { currentState, request ->
-            when (currentState) {
-                ThunderState.IDLE -> Unit
-                ThunderState.CONNECTING -> {}
+        ) { currentSocketState, request ->
+            when (currentSocketState) {
                 ThunderState.CONNECTED -> {
-                    if (isFromError && recoveryCache.hasCache()) {
+                    if (isReSubscription && recoveryCache.hasCache()) {
                         recoveryCache.get()?.let { requestSendMessage(it) }
                         recoveryCache.clear()
-                        isFromError = false
+                        isReSubscription = false
                     } else {
                         request.forEach(::requestSendMessage)
                     }
                 }
-                ThunderState.DISCONNECTING -> {}
-                ThunderState.DISCONNECTED -> {}
-                is ThunderState.ERROR -> {}
+
+                else -> Unit
             }
-        }.launchIn(scope)
+        }.launchIn(innerScope)
+    }
+
+    private suspend fun retryConnection() {
+        thunderLog("Thunder retry connection work.")
+        closeConnection()
+        delay(RETRY_CONNECTION_GAP)
+        openConnection()
+        _retryNeedFlag.update { false }
     }
 
     /**
@@ -152,20 +173,22 @@ class ThunderStateManager private constructor(
     }
 
     private lateinit var connectionJob: Job
-    private fun openConnection() {
+    private fun openConnection() = synchronized(this) {
         if (socket == null) {
-            _socketState.updateThunderState(ThunderState.CONNECTING)
             socket = webSocketCore.create()
             socket?.let { webSocket ->
                 webSocket.open()
+                thunderLog("Thunder open connection work.")
                 if (::connectionJob.isInitialized) connectionJob.cancel()
-                connectionJob = webSocket.events().onEach { _events.tryEmit(it) }.launchIn(scope)
+                connectionJob = webSocket.events().onEach { _events.tryEmit(it) }.launchIn(innerScope)
             }
         }
     }
 
-    private fun closeConnection() {
+    private fun closeConnection() = synchronized(this) {
         socket?.let {
+            thunderLog("Thunder close connection work.")
+            _socketState.update { ThunderState.ERROR() }
             it.close(1000, "shutdown")
             if (::connectionJob.isInitialized) connectionJob.cancel()
             socket = null
@@ -176,10 +199,6 @@ class ThunderStateManager private constructor(
         valveCache.requestToValve(key to message)
     }
 
-    private fun MutableStateFlow<ThunderState>.updateThunderState(state: ThunderState) {
-        _lastSocketState = getAndUpdate { state }
-    }
-
     class Factory(
         private val connectionListener: AppConnectionListener,
         private val networkStatus: NetworkConnectivityService,
@@ -197,4 +216,8 @@ class ThunderStateManager private constructor(
             )
         }
     }
+
+    companion object {
+        private const val RETRY_CONNECTION_GAP = 500L
+    }
 }

thunder/src/main/java/com/jeremy/thunder/state/ActivityState.kt

@@ -4,6 +4,8 @@ sealed interface ActivityState
 
 object Initialize: ActivityState
 
-object GetReady: ActivityState
+object Foreground: ActivityState
+
+object Background: ActivityState
 
 object ShutDown: ActivityState

thunder/src/main/java/com/jeremy/thunder/state/ThunderError.kt

@@ -1,15 +1,11 @@
 package com.jeremy.thunder.state
 
 sealed interface ThunderError {
-    data class NetworkLoss(
-        val message: String?
-    ) : ThunderError
+    object General : ThunderError
 
-    data class SocketLoss(
-        val message: String?
-    ) : ThunderError
+    object NetworkLoss : ThunderError
 
-    data class Else(
+    data class SocketLoss(
         val message: String?
     ) : ThunderError
 }

thunder/src/main/java/com/jeremy/thunder/state/ThunderState.kt

@@ -35,6 +35,6 @@ sealed interface ThunderState {
      * Error State and [ThunderError] must be defined.
     * */
     data class ERROR(
-        val error: ThunderError
+        val error: ThunderError = ThunderError.General
     ) : ThunderState
 }

thunder/src/main/java/com/jeremy/thunder/ws/WebSocket.kt

@@ -10,15 +10,15 @@ interface WebSocket {
 
     fun events(): Flow<WebSocketEvent>
 
-    //websocket 메세지 전송
+    //Websocket send message
     fun send(data: String): Boolean
 
-    //websocket 연결 종료 - code & reason
+    //Websocket Connection close - code & reason
     fun close(code: Int, reason: String)
 
     fun cancel()
 
-    //websocket 연결 오류
+    //Websocket Connection failure
     fun error(t: String)
 
     interface Factory{