Fix blank card and tracing state (EXPOSUREAPP-3823) (#1636)

due to use of `sendBlocking`.

Co-authored-by: harambasicluka <64483219+harambasicluka@users.noreply.github.com>

Corona-Warn-App/src/main/java/de/rki/coronawarnapp/nearby/modules/tracing/DefaultTracingStatus.kt

@@ -3,9 +3,10 @@ package de.rki.coronawarnapp.nearby.modules.tracing
 import com.google.android.gms.nearby.exposurenotification.ExposureNotificationClient
 import de.rki.coronawarnapp.exception.ExceptionCategory
 import de.rki.coronawarnapp.exception.reporting.report
+import de.rki.coronawarnapp.util.coroutine.AppScope
+import de.rki.coronawarnapp.util.flow.shareLatest
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.cancel
-import kotlinx.coroutines.channels.awaitClose
-import kotlinx.coroutines.channels.sendBlocking
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
@@ -23,28 +24,32 @@ import kotlin.coroutines.suspendCoroutine
 
 @Singleton
 class DefaultTracingStatus @Inject constructor(
-    private val client: ExposureNotificationClient
+    private val client: ExposureNotificationClient,
+    @AppScope val scope: CoroutineScope
 ) : TracingStatus {
 
     override val isTracingEnabled: Flow<Boolean> = callbackFlow<Boolean> {
-        var isRunning = true
-        while (isRunning && isActive) {
+        while (true) {
             try {
-                sendBlocking(pollIsEnabled())
+                send(pollIsEnabled())
             } catch (e: Exception) {
                 Timber.w(e, "ENF isEnabled failed.")
-                sendBlocking(false)
+                send(false)
                 e.report(ExceptionCategory.EXPOSURENOTIFICATION, TAG, null)
                 cancel("ENF isEnabled failed", e)
             }
+            if (!isActive) break
             delay(POLLING_DELAY_MS)
         }
-        awaitClose { isRunning = false }
     }
         .distinctUntilChanged()
         .onStart { Timber.v("isTracingEnabled FLOW start") }
         .onEach { Timber.v("isTracingEnabled FLOW emission: %b", it) }
         .onCompletion { Timber.v("isTracingEnabled FLOW completed.") }
+        .shareLatest(
+            tag = TAG,
+            scope = scope
+        )
 
     private suspend fun pollIsEnabled(): Boolean = suspendCoroutine { cont ->
         client.isEnabled

Corona-Warn-App/src/main/java/de/rki/coronawarnapp/util/BackgroundModeStatus.kt

@@ -6,8 +6,6 @@ import de.rki.coronawarnapp.util.di.AppContext
 import de.rki.coronawarnapp.util.flow.shareLatest
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.cancel
-import kotlinx.coroutines.channels.awaitClose
-import kotlinx.coroutines.channels.sendBlocking
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
@@ -23,18 +21,19 @@ class BackgroundModeStatus @Inject constructor(
     @AppScope private val appScope: CoroutineScope
 ) {
 
-    val isBackgroundRestricted: Flow<Boolean> = callbackFlow<Boolean> {
-        var isRunning = true
-        while (isRunning && isActive) {
+    val isBackgroundRestricted: Flow<Boolean?> = callbackFlow<Boolean> {
+        while (true) {
             try {
-                sendBlocking(pollIsBackgroundRestricted())
+                send(pollIsBackgroundRestricted())
             } catch (e: Exception) {
                 Timber.w(e, "isBackgroundRestricted failed.")
                 cancel("isBackgroundRestricted failed", e)
             }
+
+            if (!isActive) break
+
             delay(POLLING_DELAY_MS)
         }
-        awaitClose { isRunning = false }
     }
         .distinctUntilChanged()
         .shareLatest(
@@ -43,17 +42,18 @@ class BackgroundModeStatus @Inject constructor(
         )
 
     val isAutoModeEnabled: Flow<Boolean> = callbackFlow<Boolean> {
-        var isRunning = true
-        while (isRunning && isActive) {
+        while (true) {
             try {
-                sendBlocking(pollIsAutoMode())
+                send(pollIsAutoMode())
             } catch (e: Exception) {
                 Timber.w(e, "autoModeEnabled failed.")
                 cancel("autoModeEnabled failed", e)
             }
+
+            if (!isActive) break
+
             delay(POLLING_DELAY_MS)
         }
-        awaitClose { isRunning = false }
     }
         .distinctUntilChanged()
         .shareLatest(

Corona-Warn-App/src/main/java/de/rki/coronawarnapp/util/flow/FlowExtensions.kt

@@ -5,7 +5,7 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.combine
-import kotlinx.coroutines.flow.mapNotNull
+import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.flow.onCompletion
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
@@ -17,7 +17,7 @@ import timber.log.Timber
  * Helper method to create a new flow without suspending and without initial value
  * The flow collector will just wait for the first value
  */
-fun <T> Flow<T>.shareLatest(
+fun <T : Any> Flow<T>.shareLatest(
     tag: String? = null,
     scope: CoroutineScope,
     started: SharingStarted = SharingStarted.WhileSubscribed(replayExpirationMillis = 0)
@@ -40,7 +40,7 @@ fun <T> Flow<T>.shareLatest(
         started = started,
         initialValue = null
     )
-    .mapNotNull { it }
+    .filterNotNull()
 
 @Suppress("UNCHECKED_CAST", "LongParameterList")
 inline fun <T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, R> combine(

Corona-Warn-App/src/test/java/de/rki/coronawarnapp/nearby/modules/tracing/DefaultTracingStatusTest.kt

@@ -8,14 +8,17 @@ import io.mockk.clearAllMocks
 import io.mockk.every
 import io.mockk.impl.annotations.MockK
 import io.mockk.verify
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.take
 import kotlinx.coroutines.flow.toList
-import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import testhelpers.BaseTest
+import testhelpers.coroutines.runBlockingTest2
+import testhelpers.coroutines.test
 import testhelpers.gms.MockGMSTask
 
 class DefaultTracingStatusTest : BaseTest() {
@@ -26,7 +29,7 @@ class DefaultTracingStatusTest : BaseTest() {
     fun setup() {
         MockKAnnotations.init(this)
 
-        every { client.isEnabled } returns MockGMSTask.forValue(true)
+        every { client.isEnabled } answers { MockGMSTask.forValue(true) }
     }
 
     @AfterEach
@@ -34,33 +37,60 @@ class DefaultTracingStatusTest : BaseTest() {
         clearAllMocks()
     }
 
-    private fun createInstance(): DefaultTracingStatus = DefaultTracingStatus(
-        client = client
+    private fun createInstance(scope: CoroutineScope): DefaultTracingStatus = DefaultTracingStatus(
+        client = client,
+        scope = scope
     )
 
     @Test
-    fun `init is sideeffect free and lazy`() {
-        createInstance()
+    fun `init is sideeffect free and lazy`() = runBlockingTest2(ignoreActive = true) {
+        createInstance(scope = this)
+
+        advanceUntilIdle()
+
         verify { client wasNot Called }
     }
 
     @Test
-    fun `state emission works`() = runBlockingTest {
-        val instance = createInstance()
+    fun `state emission works`() = runBlockingTest2(ignoreActive = true) {
+        val instance = createInstance(scope = this)
         instance.isTracingEnabled.first() shouldBe true
     }
 
     @Test
-    fun `state is updated and polling stops on collection stop`() = runBlockingTest {
+    fun `state is updated and polling stops on cancel`() = runBlockingTest2(ignoreActive = true) {
         every { client.isEnabled } returnsMany listOf(
             true, false, true, false, true, false, true
         ).map { MockGMSTask.forValue(it) }
 
-        val instance = createInstance()
+        val instance = createInstance(scope = this)
 
         instance.isTracingEnabled.take(6).toList() shouldBe listOf(
             true, false, true, false, true, false
         )
         verify(exactly = 6) { client.isEnabled }
     }
+
+    @Test
+    fun `subscriptions are shared but not cached`() = runBlockingTest2(ignoreActive = true) {
+        val instance = createInstance(scope = this)
+
+        val collector1 = instance.isTracingEnabled.test(tag = "1", startOnScope = this)
+        val collector2 = instance.isTracingEnabled.test(tag = "2", startOnScope = this)
+
+        delay(500)
+
+        collector1.latestValue shouldBe true
+        collector2.latestValue shouldBe true
+
+        collector1.cancel()
+        collector2.cancel()
+
+        advanceUntilIdle()
+
+        verify(exactly = 1) { client.isEnabled }
+
+        every { client.isEnabled } answers { MockGMSTask.forValue(false) }
+        instance.isTracingEnabled.first() shouldBe false
+    }
 }

Corona-Warn-App/src/test/java/de/rki/coronawarnapp/util/BackgroundModeStatusTest.kt

@@ -10,18 +10,18 @@ import io.mockk.impl.annotations.MockK
 import io.mockk.mockkObject
 import io.mockk.verify
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
-import kotlinx.coroutines.test.TestCoroutineScope
-import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import testhelpers.BaseTest
+import testhelpers.coroutines.runBlockingTest2
+import testhelpers.coroutines.test
 
 class BackgroundModeStatusTest : BaseTest() {
 
     @MockK lateinit var context: Context
-    private val scope: CoroutineScope = TestCoroutineScope()
 
     @BeforeEach
     fun setup() {
@@ -34,23 +34,21 @@ class BackgroundModeStatusTest : BaseTest() {
         clearAllMocks()
     }
 
-    private fun createInstance(): BackgroundModeStatus = BackgroundModeStatus(
+    private fun createInstance(scope: CoroutineScope): BackgroundModeStatus = BackgroundModeStatus(
         context = context,
         appScope = scope
     )
 
     @Test
-    fun `init is sideeffect free and lazy`() {
-        createInstance()
+    fun `init is sideeffect free and lazy`() = runBlockingTest2(ignoreActive = true) {
+        createInstance(scope = this)
         verify { context wasNot Called }
     }
 
     @Test
-    fun isAutoModeEnabled() = runBlockingTest {
-        every { ConnectivityHelper.autoModeEnabled(any()) } returnsMany listOf(
-            true, false, true, false
-        )
-        createInstance().apply {
+    fun isAutoModeEnabled() = runBlockingTest2(ignoreActive = true) {
+        every { ConnectivityHelper.autoModeEnabled(any()) } returnsMany listOf(true, false, true, false)
+        createInstance(scope = this).apply {
             isAutoModeEnabled.first() shouldBe true
             isAutoModeEnabled.first() shouldBe false
             isAutoModeEnabled.first() shouldBe true
@@ -58,14 +56,60 @@ class BackgroundModeStatusTest : BaseTest() {
     }
 
     @Test
-    fun isBackgroundRestricted() = runBlockingTest {
-        every { ConnectivityHelper.isBackgroundRestricted(any()) } returnsMany listOf(
-            false, true, false
-        )
-        createInstance().apply {
+    fun `isAutoModeEnabled is shared but not cached`() = runBlockingTest2(ignoreActive = true) {
+        every { ConnectivityHelper.autoModeEnabled(any()) } returnsMany listOf(true, false, true, false)
+
+        val instance = createInstance(scope = this)
+
+        val collector1 = instance.isAutoModeEnabled.test(tag = "1", startOnScope = this)
+        val collector2 = instance.isAutoModeEnabled.test(tag = "2", startOnScope = this)
+
+        delay(500)
+
+        collector1.latestValue shouldBe true
+        collector2.latestValue shouldBe true
+
+        collector1.cancel()
+        collector2.cancel()
+
+        advanceUntilIdle()
+
+        verify(exactly = 1) { ConnectivityHelper.autoModeEnabled(any()) }
+
+        instance.isAutoModeEnabled.first() shouldBe false
+    }
+
+    @Test
+    fun isBackgroundRestricted() = runBlockingTest2(ignoreActive = true) {
+        every { ConnectivityHelper.isBackgroundRestricted(any()) } returnsMany listOf(false, true, false)
+        createInstance(scope = this).apply {
             isBackgroundRestricted.first() shouldBe false
             isBackgroundRestricted.first() shouldBe true
             isBackgroundRestricted.first() shouldBe false
         }
     }
+
+    @Test
+    fun `isBackgroundRestricted is shared but not cached`() = runBlockingTest2(ignoreActive = true) {
+        every { ConnectivityHelper.isBackgroundRestricted(any()) } returnsMany listOf(true, false, true, false)
+
+        val instance = createInstance(scope = this)
+
+        val collector1 = instance.isBackgroundRestricted.test(tag = "1", startOnScope = this)
+        val collector2 = instance.isBackgroundRestricted.test(tag = "2", startOnScope = this)
+
+        delay(500)
+
+        collector1.latestValue shouldBe true
+        collector2.latestValue shouldBe true
+
+        collector1.cancel()
+        collector2.cancel()
+
+        advanceUntilIdle()
+
+        verify(exactly = 1) { ConnectivityHelper.isBackgroundRestricted(any()) }
+
+        instance.isBackgroundRestricted.first() shouldBe false
+    }
 }