Change state update mechanism

Author: aleksandar-apostolov

stream-android-core/src/main/java/io/getstream/android/core/api/StreamClient.kt

@@ -25,7 +25,7 @@ import io.getstream.android.core.api.model.config.StreamHttpConfig
 import io.getstream.android.core.api.model.config.StreamSocketConfig
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
-import io.getstream.android.core.api.model.connection.network.StreamNetworkInfo
+import io.getstream.android.core.api.model.connection.network.StreamNetworkState
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.model.value.StreamHttpClientInfoHeader
 import io.getstream.android.core.api.model.value.StreamUserId
@@ -117,7 +117,7 @@ public interface StreamClient {
      * - Hot & conflated: new collectors receive the latest value immediately.
      * - `null` if no network is available.
      */
-    @StreamInternalApi public val networkInfo: StateFlow<StreamNetworkInfo.Snapshot?>
+    @StreamInternalApi public val networkState: StateFlow<StreamNetworkState>
 
     /**
      * Establishes a connection for the current user.
@@ -291,6 +291,7 @@ public fun StreamClient(
         serialQueue = serialQueue,
         connectionIdHolder = connectionIdHolder,
         logger = clientLogger,
+        mutableNetworkState = MutableStateFlow(StreamNetworkState.Unknown),
         mutableConnectionState = MutableStateFlow(StreamConnectionState.Idle),
         subscriptionManager = clientSubscriptionManager,
         networkMonitor = networkMonitor,

stream-android-core/src/main/java/io/getstream/android/core/api/model/connection/network/StreamNetworkState.kt

@@ -0,0 +1,63 @@
+package io.getstream.android.core.api.model.connection.network
+
+import io.getstream.android.core.annotations.StreamInternalApi
+
+@StreamInternalApi
+public sealed class StreamNetworkState {
+
+    /**
+     * Signals that the platform reported a permanent loss of network connectivity.
+     *
+     * This state mirrors the `ConnectivityManager.NetworkCallback.onUnavailable` callback, which
+     * indicates no viable network path exists. Applications should back off from network work and
+     * surface an offline UI until a different state is received.
+     *
+     * ### Example
+     * ```kotlin
+     * when (state) {
+     *     StreamNetworkState.Unavailable -> showOfflineBanner("No connection available")
+     *     else -> hideOfflineBanner()
+     * }
+     * ```
+     */
+    public data object Unavailable : StreamNetworkState()
+
+    /**
+     * Represents the initial, indeterminate state before any network callbacks have fired.
+     *
+     * Use this as a cue to defer UI decisions until more definitive information arrives. The state
+     * will transition to one of the other variants once the monitor observes connectivity events.
+     */
+    public data object Unknown : StreamNetworkState()
+
+    /**
+     * Indicates that a network was previously tracked but has been lost.
+     *
+     * This corresponds to `ConnectivityManager.NetworkCallback.onLost`. Stream monitors emit this
+     * when the active network disconnects but the system may still attempt reconnection, so you can
+     * show transient offline messaging or pause network-heavy tasks.
+     */
+    public data object Disconnected : StreamNetworkState()
+
+    /**
+     * A network path is currently active and considered connected.
+     *
+     * This state maps to `ConnectivityManager.NetworkCallback.onAvailable` and carries the most
+     * recent [StreamNetworkInfo.Snapshot], allowing callers to inspect transports, metering, or
+     * other network characteristics before resuming work.
+     *
+     * ### Example
+     * ```kotlin
+     * when (state) {
+     *     is StreamNetworkState.Available ->
+     *         logger.i { "Connected via ${state.snapshot?.transports}" }
+     *     StreamNetworkState.Disconnected -> logger.w { "Network dropped" }
+     *     StreamNetworkState.Unavailable -> logger.e { "No connection" }
+     *     StreamNetworkState.Unknown -> logger.d { "Awaiting first update" }
+     * }
+     * ```
+     *
+     * @property snapshot Latest network snapshot, or `null` if collection failed.
+     */
+    public data class Available(val snapshot: StreamNetworkInfo.Snapshot?) : StreamNetworkState()
+}

stream-android-core/src/main/java/io/getstream/android/core/api/socket/listeners/StreamClientListener.kt

@@ -17,6 +17,8 @@ package io.getstream.android.core.api.socket.listeners
 
 import io.getstream.android.core.annotations.StreamInternalApi
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.connection.network.StreamNetworkInfo
+import io.getstream.android.core.api.model.connection.network.StreamNetworkState
 
 /**
  * Listener interface for Feeds socket events.
@@ -46,4 +48,11 @@ public interface StreamClientListener {
      * @param err The error that occurred.
      */
     public fun onError(err: Throwable) {}
+
+    /**
+     * Called when the network connection changes.
+     *
+     * @param state The new network state.
+     */
+    public fun onNetworkState(state: StreamNetworkState) {}
 }

stream-android-core/src/main/java/io/getstream/android/core/internal/client/StreamClientImpl.kt

@@ -15,14 +15,14 @@
  */
 package io.getstream.android.core.internal.client
 
-import io.getstream.android.core.annotations.StreamInternalApi
 import io.getstream.android.core.api.StreamClient
 import io.getstream.android.core.api.authentication.StreamTokenManager
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.StreamTypedKey.Companion.randomExecutionKey
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
 import io.getstream.android.core.api.model.connection.network.StreamNetworkInfo
+import io.getstream.android.core.api.model.connection.network.StreamNetworkState
 import io.getstream.android.core.api.model.value.StreamUserId
 import io.getstream.android.core.api.observers.network.StreamNetworkMonitor
 import io.getstream.android.core.api.observers.network.StreamNetworkMonitorListener
@@ -48,6 +48,7 @@ internal class StreamClientImpl<T>(
     private val serialQueue: StreamSerialProcessingQueue,
     private val connectionIdHolder: StreamConnectionIdHolder,
     private val socketSession: StreamSocketSession<T>,
+    private var mutableNetworkState: MutableStateFlow<StreamNetworkState>,
     private val mutableConnectionState: MutableStateFlow<StreamConnectionState>,
     private val logger: StreamLogger,
     private val subscriptionManager: StreamSubscriptionManager<StreamClientListener>,
@@ -64,12 +65,9 @@ internal class StreamClientImpl<T>(
     override val connectionState: StateFlow<StreamConnectionState>
         get() = mutableConnectionState.asStateFlow()
 
-    private var internalNetworkInfo: MutableStateFlow<StreamNetworkInfo.Snapshot?> =
-        MutableStateFlow(null)
+    override val networkState: StateFlow<StreamNetworkState>
+        get() = mutableNetworkState.asStateFlow()
 
-    @StreamInternalApi
-    override val networkInfo: StateFlow<StreamNetworkInfo.Snapshot?>
-        get() = internalNetworkInfo.asStateFlow()
 
     override fun subscribe(listener: StreamClientListener): Result<StreamSubscription> =
         subscriptionManager.subscribe(listener)
@@ -135,19 +133,34 @@ internal class StreamClientImpl<T>(
                                                 logger.v {
                                                     "[connect] Network connected: $snapshot"
                                                 }
-                                                internalNetworkInfo.update(snapshot)
+                                                val state = StreamNetworkState.Available(snapshot)
+                                                mutableNetworkState.update(state)
+                                                subscriptionManager.forEach {
+                                                    it.onNetworkState(state)
+                                                }
                                             }
 
                                             override suspend fun onNetworkLost(permanent: Boolean) {
                                                 logger.v { "[connect] Network lost" }
-                                                internalNetworkInfo.update(null)
+                                                val state = if (permanent) {
+                                                    StreamNetworkState.Unavailable
+                                                } else {
+                                                    StreamNetworkState.Disconnected
+                                                }
+                                                mutableNetworkState.update(state)
+                                                subscriptionManager.forEach {
+                                                    it.onNetworkState(state)
+                                                }
                                             }
 
                                             override suspend fun onNetworkPropertiesChanged(
                                                 snapshot: StreamNetworkInfo.Snapshot
                                             ) {
                                                 logger.v { "[connect] Network changed: $snapshot" }
-                                                internalNetworkInfo.update(snapshot)
+                                                mutableNetworkState.update(StreamNetworkState.Available(snapshot))
+                                                subscriptionManager.forEach {
+                                                    it.onNetworkState(StreamNetworkState.Available(snapshot))
+                                                }
                                             }
                                         },
                                         StreamSubscriptionManager.Options(

stream-android-core/src/test/java/io/getstream/android/core/api/socket/listeners/StreamListenersDefaultImplsTest.kt

@@ -19,6 +19,7 @@ package io.getstream.android.core.api.socket.listeners
 
 import io.getstream.android.core.annotations.StreamInternalApi
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.connection.network.StreamNetworkState
 import kotlin.test.assertEquals
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.launch
@@ -50,6 +51,7 @@ internal class StreamListenersDefaultImplsTest {
         val stateChannel = Channel<StreamConnectionState>(capacity = 1)
         val eventChannel = Channel<Any>(capacity = 1)
         val errorChannel = Channel<Throwable>(capacity = 1)
+        val networkChannel = Channel<StreamNetworkState>(capacity = 1)
 
         val listener =
             object : StreamClientListener {
@@ -64,6 +66,10 @@ internal class StreamListenersDefaultImplsTest {
                 override fun onError(err: Throwable) {
                     errorChannel.trySend(err)
                 }
+
+                override fun onNetworkState(state: StreamNetworkState) {
+                    networkChannel.trySend(state)
+                }
             }
 
         val state = StreamConnectionState.Connecting.Opening("user")