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

aws-datastore/src/main/java/com/amplifyframework/datastore/extensions/NetworkExtensions.kt

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 com.amplifyframework.datastore.extensions
+
+import android.net.ConnectivityManager
+import android.net.NetworkCapabilities
+import android.os.Build
+import androidx.annotation.VisibleForTesting
+
+// Return network capabilities based on Connectivity Manager
+@VisibleForTesting(otherwise = VisibleForTesting.PRIVATE)
+fun ConnectivityManager.networkCapabilitiesOrNull() = try {
+    getNetworkCapabilities(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.
+    null
+}
+
+// Return whether internet is reachable based on network capabilities.
+fun NetworkCapabilities?.isInternetReachable() = when {
+    this == null -> false
+    hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) -> true
+    hasTransport(NetworkCapabilities.TRANSPORT_WIFI) -> true
+    hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) -> true
+    else -> false
+}
+
+// Check whether network is available based on connectivity 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
+fun ConnectivityManager.isNetworkAvailable(): Boolean =
+    if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
+        networkCapabilitiesOrNull()?.isInternetReachable() ?: false
+    } else {
+        activeNetworkInfo?.isConnected ?: false
+    }

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

@@ -19,9 +19,13 @@ import android.content.Context
 import android.net.ConnectivityManager
 import android.net.ConnectivityManager.NetworkCallback
 import android.net.Network
+import android.net.NetworkCapabilities
 import androidx.annotation.VisibleForTesting
 import com.amplifyframework.datastore.DataStoreException
+import com.amplifyframework.datastore.extensions.isInternetReachable
+import com.amplifyframework.datastore.extensions.isNetworkAvailable
 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,39 +87,64 @@ private class ReachabilityMonitorImpl constructor(val schedulerProvider: Schedul
         }
         return subject.subscribeOn(schedulerProvider.io())
     }
+
+    private inner class ConnectivityNetworkCallback(private val emitter: ObservableEmitter<Boolean>) : NetworkCallback() {
+        private var currentNetworkCapabilities: NetworkCapabilities? = null
+
+        override fun onAvailable(network: Network) {
+            // in recent Android SDK version, onCapabilitiesChanged is triggered before onAvailable
+            // we should make use of network capabilities if it was updated before
+            currentNetworkCapabilities?.let {
+                emitter.onNext(it.isInternetReachable())
+            } ?: run {
+                connectivityProvider?.let {
+                    emitter.onNext(it.hasActiveNetwork)
+                }
+            }
+        }
+
+        override fun onLost(network: Network) {
+            currentNetworkCapabilities = null
+            emitter.onNext(false)
+        }
+
+        override fun onUnavailable() {
+            currentNetworkCapabilities = null
+            emitter.onNext(false)
+        }
+
+        override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) {
+            currentNetworkCapabilities = networkCapabilities
+            emitter.onNext(networkCapabilities.isInternetReachable())
+        }
+    }
 }
 
 /**
  * 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?.isNetworkAvailable() ?: 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) }
-            ?: run {
-                throw DataStoreException(
-                    "ConnectivityManager not available",
-                    "No recovery suggestion is available"
-                )
-            }
+        connectivityManager?.registerDefaultNetworkCallback(callback)
+            ?: throw DataStoreException(
+                "ConnectivityManager not available",
+                "No recovery suggestion is available"
+            )
     }
 }

aws-datastore/src/test/java/com/amplifyframework/datastore/extensions/NetworkExtensionsTest.kt

@@ -0,0 +1,62 @@
+package com.amplifyframework.datastore.extensions
+
+import android.net.ConnectivityManager
+import android.net.NetworkCapabilities
+import io.mockk.every
+import io.mockk.mockk
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Assert.assertFalse
+import org.junit.Test
+
+class NetworkCapabilitiesUtilTest {
+
+    @Test
+    fun testGetNetworkCapabilitiesSecurityException() {
+        val mockConnectivityManager = mockk<ConnectivityManager> {
+            every { activeNetwork } throws SecurityException()
+        }
+
+        val networkCapabilities = mockConnectivityManager.networkCapabilitiesOrNull()
+        assertNull(networkCapabilities)
+    }
+
+    @Test
+    fun testGetNetworkCapabilities() {
+        val expectedNetworkCapabilities = mockk<NetworkCapabilities>()
+        val mockConnectivityManager = mockk<ConnectivityManager> {
+            every { getNetworkCapabilities(any()) } returns expectedNetworkCapabilities
+        }
+
+        val networkCapabilities = mockConnectivityManager.networkCapabilitiesOrNull()
+
+        assertEquals(expectedNetworkCapabilities, networkCapabilities)
+    }
+
+    @Test
+    fun testIsInternetReachable() {
+        val networkCapabilitiesWithCellular = mockk<NetworkCapabilities> {
+            every { hasTransport(NetworkCapabilities.TRANSPORT_WIFI) } returns false
+            every { hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) } returns true
+            every { hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) } returns false
+        }
+
+        val networkCapabilitiesWithWifi = mockk<NetworkCapabilities> {
+            every { hasTransport(NetworkCapabilities.TRANSPORT_WIFI) } returns true
+            every { hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) } returns false
+            every { hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) } returns false
+        }
+
+        val networkCapabilitiesWithEthernet = mockk<NetworkCapabilities> {
+            every { hasTransport(NetworkCapabilities.TRANSPORT_WIFI) } returns false
+            every { hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) } returns false
+            every { hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) } returns true
+        }
+
+        assertTrue(networkCapabilitiesWithCellular.isInternetReachable())
+        assertTrue(networkCapabilitiesWithWifi.isInternetReachable())
+        assertTrue(networkCapabilitiesWithEthernet.isInternetReachable())
+        assertFalse(null.isInternetReachable())
+    }
+}

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

@@ -18,15 +18,17 @@ 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.mockk.every
+import io.mockk.mockk
 import io.reactivex.rxjava3.core.BackpressureStrategy
 import io.reactivex.rxjava3.schedulers.TestScheduler
 import io.reactivex.rxjava3.subscribers.TestSubscriber
 import java.util.concurrent.TimeUnit
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
 import org.junit.Test
-import org.mockito.Mockito.mock
 
 class ReachabilityMonitorTest {
 
@@ -56,7 +58,7 @@ class ReachabilityMonitorTest {
             }
         }
 
-        val mockContext = mock(Context::class.java)
+        val mockContext = mockk<Context>(relaxed = true)
         // TestScheduler allows the virtual time to be advanced by exact amounts, to allow for repeatable tests
         val testScheduler = TestScheduler()
         val reachabilityMonitor = ReachabilityMonitor.createForTesting(TestSchedulerProvider(testScheduler))
@@ -69,18 +71,25 @@ class ReachabilityMonitorTest {
             .toFlowable(BackpressureStrategy.BUFFER)
             .subscribe(testSubscriber)
 
-        val network = mock(Network::class.java)
+        val network = mockk<Network>(relaxed = true)
+        val networkCapabilities = mockk<NetworkCapabilities>()
+        every { networkCapabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) } returns 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)
 
@@ -108,7 +117,7 @@ class ReachabilityMonitorTest {
             }
         }
 
-        val mockContext = mock(Context::class.java)
+        val mockContext = mockk<Context>(relaxed = true)
         // TestScheduler allows the virtual time to be advanced by exact amounts, to allow for repeatable tests
         val testScheduler = TestScheduler()
         val reachabilityMonitor = ReachabilityMonitor.createForTesting(TestSchedulerProvider(testScheduler))
@@ -121,10 +130,13 @@ class ReachabilityMonitorTest {
             .toFlowable(BackpressureStrategy.BUFFER)
             .subscribe(testSubscriber)
 
-        val network = mock(Network::class.java)
+        val network = mockk<Network>(relaxed = true)
+        val networkCapabilities = mockk<NetworkCapabilities>()
+        every { networkCapabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) } returns 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 +158,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
@@ -197,11 +211,11 @@ class ReachabilityMonitorTest {
         // TestScheduler allows the virtual time to be advanced by exact amounts, to allow for repeatable tests
         val testScheduler = TestScheduler()
         val reachabilityMonitor = ReachabilityMonitor.createForTesting(TestSchedulerProvider(testScheduler))
-        val mockContext = mock(Context::class.java)
+        val mockContext = mockk<Context>(relaxed = true)
         reachabilityMonitor.configure(mockContext, connectivityProvider)
 
         reachabilityMonitor.getObservable().subscribe()
-        val network = mock(Network::class.java)
+        val network = mockk<Network>(relaxed = true)
         // Should provide initial network state (true) upon subscription (after debounce)
         testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
         networkCallback!!.onAvailable(network)