Improve connectivity listener online detection

Fixes incorrect filter for networkEvents. Also adds debouncing to
offline state when roaming to cellular from WiFi, this would otherwise
cause a full reconnection.

3 files changed, 77 insertions, 89 deletions

diff --git a/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt b/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt
index 863fc32251..bf94c80778 100644
--- a/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt
+++ b/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt
@@ -1,26 +1,9 @@
 package net.mullvad.mullvadvpn.lib.common.util
 
-import kotlin.time.Duration
-import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.debounce
 import kotlinx.coroutines.flow.firstOrNull
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.withIndex
 import kotlinx.coroutines.withTimeoutOrNull
 
 suspend fun <T> Flow<T>.firstOrNullWithTimeout(timeMillis: Long): T? {
     return withTimeoutOrNull(timeMillis) { firstOrNull() }
 }
-
-@OptIn(FlowPreview::class)
-fun <T> Flow<T>.debounceFirst(timeout: Duration): Flow<T> =
-    withIndex()
-        .debounce {
-            if (it.index == 0) {
-                timeout
-            } else {
-                Duration.ZERO
-            }
-        }
-        .map { it.value }
diff --git a/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt b/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt
index 3505e4496b..c918d762ff 100644
--- a/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt
+++ b/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt
@@ -2,35 +2,23 @@ package net.mullvad.talpid
 
 import android.net.ConnectivityManager
 import android.net.LinkProperties
-import android.net.Network
-import android.net.NetworkCapabilities
-import android.net.NetworkRequest
-import co.touchlab.kermit.Logger
 import java.net.InetAddress
 import kotlin.collections.ArrayList
-import kotlin.time.Duration.Companion.seconds
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.distinctUntilChanged
-import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.receiveAsFlow
-import kotlinx.coroutines.flow.scan
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
-import net.mullvad.mullvadvpn.lib.common.util.debounceFirst
 import net.mullvad.talpid.model.NetworkState
-import net.mullvad.talpid.util.NetworkEvent
 import net.mullvad.talpid.util.RawNetworkState
 import net.mullvad.talpid.util.defaultRawNetworkStateFlow
-import net.mullvad.talpid.util.networkEvents
+import net.mullvad.talpid.util.hasInternetConnectivity
 
 class ConnectivityListener(private val connectivityManager: ConnectivityManager) {
     private lateinit var _isConnected: StateFlow<Boolean>
@@ -64,7 +52,8 @@ class ConnectivityListener(private val connectivityManager: ConnectivityManager)
         }
 
         _isConnected =
-            hasInternetConnectivity()
+            connectivityManager
+                .hasInternetConnectivity()
                 .onEach { notifyConnectivityChange(it) }
                 .stateIn(
                     scope,
@@ -84,64 +73,6 @@ class ConnectivityListener(private val connectivityManager: ConnectivityManager)
     private fun LinkProperties.dnsServersWithoutFallback(): List<InetAddress> =
         dnsServers.filter { it.hostAddress != TalpidVpnService.FALLBACK_DUMMY_DNS_SERVER }
 
-    private val nonVPNInternetNetworksRequest =
-        NetworkRequest.Builder()
-            .addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
-            .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
-            .build()
-
-    /**
-     * Return a flow notifying us if we have internet connectivity. Initial state will be taken from
-     * `allNetworks` and then updated when network events occur. Important to note that
-     * `allNetworks` may return a network that we never get updates from if turned off at the moment
-     * of the initial query.
-     */
-    private fun hasInternetConnectivity(): Flow<Boolean> {
-        return connectivityManager
-            .networkEvents(nonVPNInternetNetworksRequest)
-            .filter { it is NetworkEvent.Lost || it is NetworkEvent.CapabilitiesChanged }
-            .scan(emptySet<Network>()) { networks, event ->
-                when (event) {
-                    is NetworkEvent.Lost -> networks - event.network
-                    is NetworkEvent.Available -> networks + event.network
-                    else -> networks // Should never happen
-                }.also { Logger.d("Networks: $it") }
-            }
-            // NetworkEvents are slow, can several 100 millis to arrive. If we are online, we don't
-            // want to emit a false offline with the initial accumulator, so we wait a bit before
-            // emitting, and rely on `networksWithInternetConnectivity`.
-            //
-            // Also if our initial state was "online", but it just got turned off we might not see
-            // any updates for this network even though we already were registered for updated, and
-            // thus we can't drop initial value accumulator value.
-            .debounceFirst(1.seconds)
-            .onStart {
-                // We should not use this as initial state in scan, because it may contain networks
-                // that won't be included in `networkEvents` updates.
-                emit(
-                    connectivityManager.networksWithInternetConnectivity().also {
-                        Logger.d("Networks (Initial): $it")
-                    }
-                )
-            }
-            .map { it.isNotEmpty() }
-            .distinctUntilChanged()
-    }
-
-    @Suppress("DEPRECATION")
-    private fun ConnectivityManager.networksWithInternetConnectivity(): Set<Network> =
-        // Currently the use of `allNetworks` (which is deprecated in favor of listening to network
-        // events) is our only option because network events does not give us the initial state fast
-        // enough.
-        allNetworks
-            .filter {
-                val capabilities = getNetworkCapabilities(it) ?: return@filter false
-
-                capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) &&
-                    capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
-            }
-            .toSet()
-
     private fun RawNetworkState.toNetworkState(): NetworkState =
         NetworkState(
             network.networkHandle,
diff --git a/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/util/ConnectivityManagerUtil.kt b/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/util/ConnectivityManagerUtil.kt
index f3297a995e..89ddf425f5 100644
--- a/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/util/ConnectivityManagerUtil.kt
+++ b/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/util/ConnectivityManagerUtil.kt
@@ -6,12 +6,22 @@ import android.net.LinkProperties
 import android.net.Network
 import android.net.NetworkCapabilities
 import android.net.NetworkRequest
+import co.touchlab.kermit.Logger
+import kotlin.time.Duration.Companion.milliseconds
+import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.flow.debounce
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.mapNotNull
+import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.scan
 
+private val CONNECTIVITY_DEBOUNCE = 300.milliseconds
+
 internal fun ConnectivityManager.defaultNetworkEvents(): Flow<NetworkEvent> = callbackFlow {
     val callback =
         object : NetworkCallback() {
@@ -150,3 +160,67 @@ internal data class RawNetworkState(
     val blockedStatus: Boolean = false,
     val maxMsToLive: Int? = null,
 )
+
+private val nonVPNInternetNetworksRequest =
+    NetworkRequest.Builder()
+        .addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
+        .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+        .build()
+
+private sealed interface InternalConnectivityEvent {
+    data class Available(val network: Network) : InternalConnectivityEvent
+
+    data class Lost(val network: Network) : InternalConnectivityEvent
+}
+
+/**
+ * Return a flow notifying us if we have internet connectivity. Initial state will be taken from
+ * `allNetworks` and then updated when network events occur. Important to note that `allNetworks`
+ * may return a network that we never get updates from if turned off at the moment of the initial
+ * query.
+ */
+@OptIn(FlowPreview::class)
+fun ConnectivityManager.hasInternetConnectivity(): Flow<Boolean> =
+    networkEvents(nonVPNInternetNetworksRequest)
+        .mapNotNull {
+            when (it) {
+                is NetworkEvent.Available -> InternalConnectivityEvent.Available(it.network)
+                is NetworkEvent.Lost -> InternalConnectivityEvent.Lost(it.network)
+                else -> null
+            }
+        }
+        .scan(emptySet<Network>()) { networks, event ->
+            when (event) {
+                is InternalConnectivityEvent.Lost -> networks - event.network
+                is InternalConnectivityEvent.Available -> networks + event.network
+            }.also { Logger.d("Networks: $it") }
+        }
+        // NetworkEvents are slow, can several 100 millis to arrive. If we are online, we don't
+        // want to emit a false offline with the initial accumulator, so we wait a bit before
+        // emitting, and rely on `networksWithInternetConnectivity`.
+        //
+        // Also if our initial state was "online", but it just got turned off we might not see
+        // any updates for this network even though we already were registered for updated, and
+        // thus we can't drop initial value accumulator value.
+        .debounce(CONNECTIVITY_DEBOUNCE)
+        .onStart {
+            // We should not use this as initial state in scan, because it may contain networks
+            // that won't be included in `networkEvents` updates.
+            emit(networksWithInternetConnectivity().also { Logger.d("Networks (Initial): $it") })
+        }
+        .map { it.isNotEmpty() }
+        .distinctUntilChanged()
+
+@Suppress("DEPRECATION")
+fun ConnectivityManager.networksWithInternetConnectivity(): Set<Network> =
+    // Currently the use of `allNetworks` (which is deprecated in favor of listening to network
+    // events) is our only option because network events does not give us the initial state fast
+    // enough.
+    allNetworks
+        .filter {
+            val capabilities = getNetworkCapabilities(it) ?: return@filter false
+
+            capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) &&
+                capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
+        }
+        .toSet()