GetStream · aleksandar-apostolov · Aug 25, 2026 · Aug 25, 2026 · Aug 25, 2026 · Aug 26, 2026
@@ -42,7 +42,10 @@ internal class StreamLifecycleObserver(
         withContext(DispatcherProvider.Main) {
             handlers = handlers + handler
             if (isObserving.compareAndSet(false, true)) {
-                recurringResumeEvent = false
+                // addObserver replays ON_RESUME when the owner is already resumed, and that replay
+                // is the only event worth ignoring. Subscribing from a non-resumed process gets no
+                // replay, so the next ON_RESUME is a real foregrounding and must be delivered.
+                recurringResumeEvent = !lifecycle.currentState.isAtLeast(Lifecycle.State.RESUMED)
                 lifecycle.addObserver(this@StreamLifecycleObserver)
                 logger.v { "[observe] subscribed" }
             }

@@ -47,32 +47,38 @@ internal class NetworkStateProvider(
 
         override fun onLost(network: Network) {
             availableNetworks.remove(network)
-            notifyListenersIfNetworkStateChanged()
             if (availableNetworks.isEmpty()) {
-                // No available networks, notify listeners about disconnection
-                listeners.onDisconnected()
+                // No available networks, the capability read may still lag behind
+                setConnected(false)
+            } else {
+                notifyListenersIfNetworkStateChanged()
             }
         }
     }
 
     @Volatile
-    private var isConnected: Boolean = isConnected()
+    private var lastKnownConnected: Boolean = queryConnectivity()
 
     @Volatile
     private var listeners: Set<NetworkStateListener> = setOf()
 
     private val isRegistered: AtomicBoolean = AtomicBoolean(false)
 
     private fun notifyListenersIfNetworkStateChanged() {
-        val isNowConnected = isConnected()
-        if (!isConnected && isNowConnected) {
-            logger.i { "Network connected." }
-            isConnected = true
-            listeners.onConnected()
-        } else if (isConnected && !isNowConnected) {
-            logger.i { "Network disconnected." }
-            isConnected = false
-            listeners.onDisconnected()
+        setConnected(queryConnectivity())
+    }
+
+    private fun setConnected(isNowConnected: Boolean) {
+        synchronized(lock) {
+            if (lastKnownConnected == isNowConnected) return
+            lastKnownConnected = isNowConnected
+            if (isNowConnected) {
+                logger.i { "Network connected." }
+                listeners.onConnected()
+            } else {
+                logger.i { "Network disconnected." }
+                listeners.onDisconnected()
+            }
         }
     }
 
@@ -88,7 +94,24 @@ internal class NetworkStateProvider(
         }
     }
 
-    fun isConnected(): Boolean {
+    /**
+     * Reports whether the network is currently usable.
+     *
+     * Only a `false` answer is recorded. A caller that concludes "no network" from this must not
+     * leave the last known state stale, or the following reconnection is not seen as a transition
+     * and listeners are never told the network came back.
+     *
+     * A `true` answer is deliberately not recorded: doing so would consume the very transition the
+     * listeners are waiting for, whenever this is read after the network returns but before the
+     * system callback lands.
+     */
+    fun isConnected(): Boolean = queryConnectivity().also { connected ->
+        if (!connected) {
+            synchronized(lock) { lastKnownConnected = false }
+        }
+    }
+
+    private fun queryConnectivity(): Boolean {
         return if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
             runCatching {
                 connectivityManager.run {

@@ -208,6 +208,7 @@ internal class ChatSocketStateService(initialState: State = State.Disconnected.S
                 onEvent<Event.RequiredDisconnection> { State.Disconnected.DisconnectedByRequest }
                 onEvent<Event.Stop> { State.Disconnected.Stopped }
                 onEvent<Event.NetworkAvailable> { State.RestartConnection(RestartReason.NETWORK_AVAILABLE) }
+                onEvent<Event.Resume> { State.RestartConnection(RestartReason.LIFECYCLE_RESUME) }
             }
 
             state<State.Disconnected.WebSocketEventLost> {

@@ -0,0 +1,77 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-chat-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.chat.android.client
+
+import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.testing.TestLifecycleOwner
+import io.getstream.chat.android.test.TestCoroutineExtension
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.RegisterExtension
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
+import org.mockito.kotlin.verifyBlocking
+
+internal class StreamLifecycleObserverTest {
+
+    private companion object {
+        @JvmField
+        @RegisterExtension
+        val testCoroutines = TestCoroutineExtension()
+    }
+
+    /**
+     * Regression test. Subscribing from a process that is not resumed gets no replayed ON_RESUME,
+     * so the next one is a real foregrounding and has to reach the handlers. Before the fix it was
+     * swallowed together with the replay, and a socket waiting on a resume was never woken.
+     */
+    @Test
+    fun `when subscribed while not resumed, the following resume is delivered`() = runTest(testCoroutines.dispatcher) {
+        val owner = TestLifecycleOwner(Lifecycle.State.CREATED, testCoroutines.dispatcher)
+        val observer = StreamLifecycleObserver(testCoroutines.scope, owner.lifecycle)
+        val handler: LifecycleHandler = mock()
+
+        observer.observe(handler)
+        owner.setCurrentState(Lifecycle.State.RESUMED)
+
+        verifyBlocking(handler) { resume() }
+    }
+
+    @Test
+    fun `when subscribed while already resumed, the replayed resume is ignored`() = runTest(testCoroutines.dispatcher) {
+        val owner = TestLifecycleOwner(Lifecycle.State.RESUMED, testCoroutines.dispatcher)
+        val observer = StreamLifecycleObserver(testCoroutines.scope, owner.lifecycle)
+        val handler: LifecycleHandler = mock()
+
+        observer.observe(handler)
+
+        verifyBlocking(handler, never()) { resume() }
+    }
+
+    @Test
+    fun `when resumed again after the first delivery, the resume is still delivered`() = runTest(testCoroutines.dispatcher) {
+        val owner = TestLifecycleOwner(Lifecycle.State.RESUMED, testCoroutines.dispatcher)
+        val observer = StreamLifecycleObserver(testCoroutines.scope, owner.lifecycle)
+        val handler: LifecycleHandler = mock()
+        observer.observe(handler)
+
+        owner.setCurrentState(Lifecycle.State.CREATED)
+        owner.setCurrentState(Lifecycle.State.RESUMED)
+
+        verifyBlocking(handler) { resume() }
+    }
+}
@@ -0,0 +1,161 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-chat-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.chat.android.client.network
+
+import android.net.ConnectivityManager
+import android.net.Network
+import android.net.NetworkCapabilities
+import android.net.NetworkRequest
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runTest
+import org.amshove.kluent.`should be equal to`
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.mockito.kotlin.any
+import org.mockito.kotlin.anyOrNull
+import org.mockito.kotlin.argumentCaptor
+import org.mockito.kotlin.doReturn
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
+import org.mockito.kotlin.verify
+import org.mockito.kotlin.verifyBlocking
+import org.mockito.kotlin.whenever
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+
+@OptIn(ExperimentalCoroutinesApi::class)
+@RunWith(RobolectricTestRunner::class)
+@Config(sdk = [33])
+internal class NetworkStateProviderTest {
+
+    private val connectivityManager: ConnectivityManager = mock()
+    private val capabilities: NetworkCapabilities = mock()
+
+    private fun givenNetworkUsable(usable: Boolean) {
+        whenever(capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)) doReturn usable
+        whenever(capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)) doReturn usable
+    }
+
+    private fun givenConnectivityManager() {
+        whenever(connectivityManager.activeNetwork) doReturn mock<Network>()
+        whenever(connectivityManager.getNetworkCapabilities(anyOrNull())) doReturn capabilities
+    }
+
+    private fun captureCallback(): ConnectivityManager.NetworkCallback {
+        val captor = argumentCaptor<ConnectivityManager.NetworkCallback>()
+        verify(connectivityManager).registerNetworkCallback(any<NetworkRequest>(), captor.capture())
+        return captor.firstValue
+    }
+
+    /**
+     * Regression test. A caller that reads [NetworkStateProvider.isConnected] while the network is
+     * momentarily unusable must not suppress the later "network is back" notification. Before the
+     * fix the cached state stayed `true`, so the return of the network was not seen as a
+     * transition and the socket was never told to reconnect.
+     */
+    @Test
+    fun `when the network is read as unusable directly, its return is still reported`() = runTest {
+        givenConnectivityManager()
+        givenNetworkUsable(true)
+        val provider = NetworkStateProvider(
+            CoroutineScope(UnconfinedTestDispatcher(testScheduler)),
+            connectivityManager,
+        )
+        val listener: NetworkStateProvider.NetworkStateListener = mock()
+        provider.subscribe(listener)
+        val callback = captureCallback()
+
+        givenNetworkUsable(false)
+        provider.isConnected() `should be equal to` false
+
+        givenNetworkUsable(true)
+        callback.onCapabilitiesChanged(mock(), capabilities)
+        advanceUntilIdle()
+
+        verifyBlocking(listener) { onConnected() }
+    }
+
+    /**
+     * Regression test. Reading [NetworkStateProvider.isConnected] after the network has returned
+     * but before the system callback lands must not consume the transition. Recording a `true`
+     * answer here would leave the callback with nothing to report, stranding the socket exactly as
+     * the stale `true` did.
+     */
+    @Test
+    fun `when the network is read as usable before the callback lands, the callback still reports it`() = runTest {
+        givenConnectivityManager()
+        givenNetworkUsable(true)
+        val provider = NetworkStateProvider(
+            CoroutineScope(UnconfinedTestDispatcher(testScheduler)),
+            connectivityManager,
+        )
+        val listener: NetworkStateProvider.NetworkStateListener = mock()
+        provider.subscribe(listener)
+        val callback = captureCallback()
+
+        givenNetworkUsable(false)
+        provider.isConnected() `should be equal to` false
+
+        givenNetworkUsable(true)
+        provider.isConnected() `should be equal to` true
+
+        callback.onCapabilitiesChanged(mock(), capabilities)
+        advanceUntilIdle()
+
+        verifyBlocking(listener) { onConnected() }
+    }
+
+    @Test
+    fun `when the network stays usable, listeners are not notified again`() = runTest {
+        givenConnectivityManager()
+        givenNetworkUsable(true)
+        val provider = NetworkStateProvider(
+            CoroutineScope(UnconfinedTestDispatcher(testScheduler)),
+            connectivityManager,
+        )
+        val listener: NetworkStateProvider.NetworkStateListener = mock()
+        provider.subscribe(listener)
+        val callback = captureCallback()
+
+        callback.onCapabilitiesChanged(mock(), capabilities)
+        advanceUntilIdle()
+
+        verifyBlocking(listener, never()) { onConnected() }
+    }
+
+    @Test
+    fun `when the last network is lost, listeners are notified of the disconnection`() = runTest {
+        givenConnectivityManager()
+        givenNetworkUsable(true)
+        val provider = NetworkStateProvider(
+            CoroutineScope(UnconfinedTestDispatcher(testScheduler)),
+            connectivityManager,
+        )
+        val listener: NetworkStateProvider.NetworkStateListener = mock()
+        provider.subscribe(listener)
+        val callback = captureCallback()
+
+        givenNetworkUsable(false)
+        callback.onLost(mock())
+        advanceUntilIdle()
+
+        verifyBlocking(listener) { onDisconnected() }
+    }
+}
@@ -413,7 +413,7 @@ internal class ChatSocketStateServiceTest {
             ),
             Arguments.of(
                 State.Disconnected.NetworkDisconnected,
-                State.Disconnected.NetworkDisconnected,
+                State.RestartConnection(RestartReason.LIFECYCLE_RESUME),
             ),
             Arguments.of(
                 State.Disconnected.WebSocketEventLost,