fix(datastore): Update network connection availability checking status logic in ReachabilityMonitor of Datastore

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/NetworkCapabilitiesUtil.kt

@@ -0,0 +1,46 @@
+package com.amplifyframework.datastore.syncengine
+
+import android.net.ConnectivityManager
+import android.net.NetworkCapabilities
+
+/**
+ * The NetworkCapabilitiesUtil provides convenient methods to check network capabilities and connection status
+ */
+object NetworkCapabilitiesUtil {
+    fun getNetworkCapabilities(connectivityManager: ConnectivityManager?): NetworkCapabilities? {
+        try {
+            return connectivityManager?.let {
+                it.getNetworkCapabilities(it.activeNetwork)
+            }
+        } catch (ignored: SecurityException) {
+            // Android 11 may throw a 'package does not belong' security exception here.
+            // Google fixed Android 14, 13 and 12 with the issue where Chaland Jean patched those versions.
+            // Android 11 is too old, so that's why we have to catch this exception here to be safe.
+            // https://android.googlesource.com/platform/frameworks/base/+/249be21013e389837f5b2beb7d36890b25ecfaaf%5E%21/
+            // We need to catch this to prevent app crash.
+        }
+        return null
+    }
+
+    fun isInternetReachable(capabilities: NetworkCapabilities?): Boolean {
+        if (capabilities == null) {
+            return false
+        }
+
+        return when {
+            capabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) -> {
+                true
+            }
+            capabilities.hasTransport(NetworkCapabilities.TRANSPORT_WIFI) -> {
+                true
+            }
+            capabilities.hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) -> {
+                true
+            }
+            capabilities.hasTransport(NetworkCapabilities.TRANSPORT_VPN) -> {
+                capabilities.linkDownstreamBandwidthKbps != 0
+            }
+            else -> false
+        }
+    }
+}

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/ReachabilityMonitor.kt

@@ -18,10 +18,14 @@ package com.amplifyframework.datastore.syncengine
 import android.content.Context
 import android.net.ConnectivityManager
 import android.net.ConnectivityManager.NetworkCallback
+import android.net.LinkProperties
 import android.net.Network
+import android.net.NetworkCapabilities
+import android.os.Build
 import androidx.annotation.VisibleForTesting
 import com.amplifyframework.datastore.DataStoreException
 import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.core.ObservableEmitter
 import io.reactivex.rxjava3.subjects.BehaviorSubject
 import java.util.concurrent.TimeUnit
 
@@ -35,7 +39,7 @@ import java.util.concurrent.TimeUnit
  * The network changes are debounced with a 250 ms delay to allow some time for one network to connect after another
  * network has disconnected (for example, wifi is lost, then cellular connects) to reduce thrashing.
  */
-public interface ReachabilityMonitor {
+interface ReachabilityMonitor {
     fun configure(context: Context)
     @VisibleForTesting
     fun configure(context: Context, connectivityProvider: ConnectivityProvider)
@@ -65,15 +69,7 @@ private class ReachabilityMonitorImpl constructor(val schedulerProvider: Schedul
         val observable = Observable.create { emitter ->
             connectivityProvider.registerDefaultNetworkCallback(
                 context,
-                object : NetworkCallback() {
-                    override fun onAvailable(network: Network) {
-                        emitter.onNext(true)
-                    }
-
-                    override fun onLost(network: Network) {
-                        emitter.onNext(false)
-                    }
-                }
+                ConnectivityNetworkCallback(emitter)
             )
             emitter.onNext(connectivityProvider.hasActiveNetwork)
         }
@@ -91,34 +87,73 @@ private class ReachabilityMonitorImpl constructor(val schedulerProvider: Schedul
         }
         return subject.subscribeOn(schedulerProvider.io())
     }
+
+    private inner class ConnectivityNetworkCallback(private val emitter: ObservableEmitter<Boolean>) : NetworkCallback() {
+        private var currentNetwork: Network? = null
+        private var currentCapabilities: NetworkCapabilities? = null
+
+        override fun onAvailable(network: Network) {
+            currentNetwork = network
+            updateAndSend()
+        }
+
+        override fun onLost(network: Network) {
+            currentNetwork = null
+            currentCapabilities = null
+            updateAndSend()
+        }
+
+        override fun onUnavailable() {
+            currentNetwork = null
+            currentCapabilities = null
+            updateAndSend()
+        }
+
+        override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) {
+            currentNetwork = network
+            currentCapabilities = networkCapabilities
+            updateAndSend()
+        }
+
+        private fun updateAndSend() {
+            emitter.onNext(NetworkCapabilitiesUtil.isInternetReachable(currentCapabilities))
+        }
+    }
 }
 
 /**
  * This interface puts an abstraction layer over ConnectivityManager. Since ConnectivityManager
  * is a concrete class created within context.getSystemService() it can't be overridden with a test
  * implementation, so this interface works around that issue.
  */
-public interface ConnectivityProvider {
+interface ConnectivityProvider {
     val hasActiveNetwork: Boolean
     fun registerDefaultNetworkCallback(context: Context, callback: NetworkCallback)
 }
 
 private class DefaultConnectivityProvider : ConnectivityProvider {
-
     private var connectivityManager: ConnectivityManager? = null
 
     override val hasActiveNetwork: Boolean
-        get() = connectivityManager?.let { it.activeNetwork != null }
-            ?: run {
-                throw DataStoreException(
-                    "ReachabilityMonitor has not been configured.",
-                    "Call ReachabilityMonitor.configure() before calling ReachabilityMonitor.getObservable()"
-                )
+        get() = connectivityManager?.let { manager ->
+            // the same logic as https://github.com/aws-amplify/amplify-android/blob/main/aws-storage-s3/src/main/java/com/amplifyframework/storage/s3/transfer/worker/BaseTransferWorker.kt#L176
+            if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
+                val networkCapabilities: NetworkCapabilities? = NetworkCapabilitiesUtil.getNetworkCapabilities(manager)
+                NetworkCapabilitiesUtil.isInternetReachable(networkCapabilities)
+            } else {
+                val activeNetworkInfo = manager.activeNetworkInfo
+                activeNetworkInfo != null && activeNetworkInfo.isConnected
             }
+        } ?: run {
+            throw DataStoreException(
+                "ReachabilityMonitor has not been configured.",
+                "Call ReachabilityMonitor.configure() before calling ReachabilityMonitor.getObservable()"
+            )
+        }
 
     override fun registerDefaultNetworkCallback(context: Context, callback: NetworkCallback) {
         connectivityManager = context.getSystemService(ConnectivityManager::class.java)
-        connectivityManager?.let { it.registerDefaultNetworkCallback(callback) }
+        connectivityManager?.registerDefaultNetworkCallback(callback)
             ?: run {
                 throw DataStoreException(
                     "ConnectivityManager not available",

aws-datastore/src/test/java/com/amplifyframework/datastore/syncengine/NetworkCapabilitiesUtilTest.kt

@@ -0,0 +1,61 @@
+package com.amplifyframework.datastore.syncengine
+
+import android.net.ConnectivityManager
+import android.net.NetworkCapabilities
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Assert.assertFalse
+import org.junit.Test
+import org.mockito.Mockito
+
+class NetworkCapabilitiesUtilTest {
+
+    @Test
+    fun testGetNetworkCapabilitiesSecurityException() {
+        val mockConnectivityManager = Mockito.mock(ConnectivityManager::class.java)
+        Mockito.`when`(mockConnectivityManager.activeNetwork).thenThrow(SecurityException())
+
+        val networkCapabilities = NetworkCapabilitiesUtil.getNetworkCapabilities(mockConnectivityManager)
+        assertNull(networkCapabilities)
+    }
+
+    @Test
+    fun testGetNetworkCapabilities() {
+        val mockConnectivityManager = Mockito.mock(ConnectivityManager::class.java)
+        val expectedNetworkCapabilities = Mockito.mock(NetworkCapabilities::class.java)
+        Mockito.`when`(mockConnectivityManager.getNetworkCapabilities(mockConnectivityManager.activeNetwork))
+            .thenReturn(expectedNetworkCapabilities)
+
+        val networkCapabilities = NetworkCapabilitiesUtil.getNetworkCapabilities(mockConnectivityManager)
+
+        assertEquals(expectedNetworkCapabilities, networkCapabilities)
+    }
+
+    @Test
+    fun testIsInternetReachable() {
+        val networkCapabilitiesWithCellular = Mockito.mock(NetworkCapabilities::class.java)
+        Mockito.`when`(networkCapabilitiesWithCellular.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR)).thenReturn(true)
+
+        val networkCapabilitiesWithWifi = Mockito.mock(NetworkCapabilities::class.java)
+        Mockito.`when`(networkCapabilitiesWithWifi.hasTransport(NetworkCapabilities.TRANSPORT_WIFI)).thenReturn(true)
+
+        val networkCapabilitiesWithEthernet = Mockito.mock(NetworkCapabilities::class.java)
+        Mockito.`when`(networkCapabilitiesWithEthernet.hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET)).thenReturn(true)
+
+        val networkCapabilitiesWithVpnAndBandwidth = Mockito.mock(NetworkCapabilities::class.java)
+        Mockito.`when`(networkCapabilitiesWithVpnAndBandwidth.hasTransport(NetworkCapabilities.TRANSPORT_VPN)).thenReturn(true)
+        Mockito.`when`(networkCapabilitiesWithVpnAndBandwidth.linkDownstreamBandwidthKbps).thenReturn(1000)
+
+        val networkCapabilitiesWithVpnNoBandwidth = Mockito.mock(NetworkCapabilities::class.java)
+        Mockito.`when`(networkCapabilitiesWithVpnNoBandwidth.hasTransport(NetworkCapabilities.TRANSPORT_VPN)).thenReturn(true)
+        Mockito.`when`(networkCapabilitiesWithVpnNoBandwidth.linkDownstreamBandwidthKbps).thenReturn(0)
+
+        assertTrue(NetworkCapabilitiesUtil.isInternetReachable(networkCapabilitiesWithCellular))
+        assertTrue(NetworkCapabilitiesUtil.isInternetReachable(networkCapabilitiesWithWifi))
+        assertTrue(NetworkCapabilitiesUtil.isInternetReachable(networkCapabilitiesWithEthernet))
+        assertTrue(NetworkCapabilitiesUtil.isInternetReachable(networkCapabilitiesWithVpnAndBandwidth))
+        assertFalse(NetworkCapabilitiesUtil.isInternetReachable(networkCapabilitiesWithVpnNoBandwidth))
+        assertFalse(NetworkCapabilitiesUtil.isInternetReachable(null))
+    }
+}

aws-datastore/src/test/java/com/amplifyframework/datastore/syncengine/ReachabilityMonitorTest.kt

@@ -18,6 +18,7 @@ package com.amplifyframework.datastore.syncengine
 import android.content.Context
 import android.net.ConnectivityManager
 import android.net.Network
+import android.net.NetworkCapabilities
 import com.amplifyframework.datastore.DataStoreException
 import io.reactivex.rxjava3.core.BackpressureStrategy
 import io.reactivex.rxjava3.schedulers.TestScheduler
@@ -27,6 +28,7 @@ import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
 import org.junit.Test
 import org.mockito.Mockito.mock
+import org.mockito.Mockito.`when`
 
 class ReachabilityMonitorTest {
 
@@ -70,17 +72,25 @@ class ReachabilityMonitorTest {
             .subscribe(testSubscriber)
 
         val network = mock(Network::class.java)
+        val networkCapabilities = mock(NetworkCapabilities::class.java)
+        `when`(networkCapabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR))
+            .thenReturn(true)
+
         // Should provide initial network state (true) upon subscription (after debounce)
         testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
         callback!!.onAvailable(network)
+        callback!!.onCapabilitiesChanged(network, networkCapabilities)
         callback!!.onAvailable(network)
+        callback!!.onCapabilitiesChanged(network, networkCapabilities)
         callback!!.onLost(network)
         // Should provide false after debounce
         testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
         callback!!.onAvailable(network)
+        callback!!.onCapabilitiesChanged(network, networkCapabilities)
         // Should provide true after debounce
         testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
         callback!!.onAvailable(network)
+        callback!!.onCapabilitiesChanged(network, networkCapabilities)
         // Should provide true after debounce
         testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
 
@@ -122,9 +132,13 @@ class ReachabilityMonitorTest {
             .subscribe(testSubscriber)
 
         val network = mock(Network::class.java)
+        val networkCapabilities = mock(NetworkCapabilities::class.java)
+        `when`(networkCapabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR))
+            .thenReturn(true)
 
         // Assert that the first value is returned
         callback!!.onAvailable(network)
+        callback!!.onCapabilitiesChanged(network, networkCapabilities)
         testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
         var result1: Boolean? = null
         val disposable1 = reachabilityMonitor.getObservable().subscribeOn(testScheduler).subscribe { result1 = it }
@@ -146,10 +160,12 @@ class ReachabilityMonitorTest {
 
         // Assert that if debouncer keeps getting restarted, value doesn't change
         callback!!.onAvailable(network)
+        callback!!.onCapabilitiesChanged(network, networkCapabilities)
         testScheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS)
         callback!!.onLost(network)
         testScheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS)
         callback!!.onAvailable(network)
+        callback!!.onCapabilitiesChanged(network, networkCapabilities)
         testScheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS)
 
         var result4: Boolean? = null