only register network callback once in reachability monitor

Author: eeatonaws

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

@@ -22,8 +22,7 @@ import android.net.Network
 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.core.ObservableOnSubscribe
+import io.reactivex.rxjava3.subjects.BehaviorSubject
 import java.util.concurrent.TimeUnit
 
 /**
@@ -54,43 +53,29 @@ public interface ReachabilityMonitor {
 }
 
 private class ReachabilityMonitorImpl constructor(val schedulerProvider: SchedulerProvider) : ReachabilityMonitor {
-    private var emitter: ObservableOnSubscribe<Boolean>? = null
-
+    private val subject = BehaviorSubject.create<Boolean>()
     override fun configure(context: Context) {
         return configure(context, DefaultConnectivityProvider())
     }
 
     override fun configure(context: Context, connectivityProvider: ConnectivityProvider) {
-        emitter = ObservableOnSubscribe { emitter ->
-            val callback = getCallback(emitter)
-            connectivityProvider.registerDefaultNetworkCallback(context, callback)
-            // Provide the current network status upon subscription.
-            emitter.onNext(connectivityProvider.hasActiveNetwork)
-        }
-    }
+        connectivityProvider.registerDefaultNetworkCallback(
+            context,
+            object : NetworkCallback() {
+                override fun onAvailable(network: Network) {
+                    subject.onNext(true)
+                }
 
-    override fun getObservable(): Observable<Boolean> {
-        emitter?.let { emitter ->
-            return Observable.create(emitter)
-                .subscribeOn(schedulerProvider.io())
-                .debounce(250, TimeUnit.MILLISECONDS, schedulerProvider.computation())
-        } ?: run {
-            throw DataStoreException(
-                "ReachabilityMonitor has not been configured.",
-                "Call ReachabilityMonitor.configure() before calling ReachabilityMonitor.getObservable()"
-            )
-        }
+                override fun onLost(network: Network) {
+                    subject.onNext(false)
+                }
+            }
+        )
     }
 
-    private fun getCallback(emitter: ObservableEmitter<Boolean>): NetworkCallback {
-        return object : NetworkCallback() {
-            override fun onAvailable(network: Network) {
-                emitter.onNext(true)
-            }
-            override fun onLost(network: Network) {
-                emitter.onNext(false)
-            }
-        }
+    override fun getObservable(): Observable<Boolean> {
+        return subject.subscribeOn(schedulerProvider.io())
+            .debounce(250, TimeUnit.MILLISECONDS, schedulerProvider.computation())
     }
 }
 

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

@@ -8,6 +8,7 @@ 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.Test
 import org.mockito.Mockito.mock
 
@@ -69,4 +70,47 @@ class ReachabilityMonitorTest {
 
         testSubscriber.assertValues(true, false, true, true)
     }
+
+    /**
+     * Test that calling getObservable() multiple times only results in the network
+     * callback being registered once.
+     */
+    @Test
+    fun testNetworkCallbackRegisteredOnce() {
+        var networkCallback: ConnectivityManager.NetworkCallback? = null
+        var numCallbacksRegistered = 0
+
+        val connectivityProvider = object : ConnectivityProvider {
+            override val hasActiveNetwork: Boolean
+                get() = run {
+                    return true
+                }
+            override fun registerDefaultNetworkCallback(
+                context: Context,
+                callback: ConnectivityManager.NetworkCallback
+            ) {
+                networkCallback = callback
+                numCallbacksRegistered += 1
+            }
+        }
+
+        // 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)
+        reachabilityMonitor.configure(mockContext, connectivityProvider)
+
+        reachabilityMonitor.getObservable().subscribe()
+        val network = mock(Network::class.java)
+        // Should provide initial network state (true) upon subscription (after debounce)
+        testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
+        networkCallback!!.onAvailable(network)
+
+        reachabilityMonitor.getObservable().subscribe()
+        testScheduler.advanceTimeBy(251, TimeUnit.MILLISECONDS)
+        networkCallback!!.onAvailable(network)
+
+        // Only 1 network callback should be registered
+        assertEquals(1, numCallbacksRegistered)
+    }
 }