[Sb refactor] Fix missed telephony callback events

This change implements a new flow operator, `coalesce`, whichs allows
for us to design a backpressure strategy in
`MobileConnectionRepositoryImpl` such that we never drop a
`TelephonyCallback` event if it is the only one of its type.

In practice, this means that if we have a late subscriber to a field
which is backed by the `callbackEvents` flow, we'll never miss data that
would otherwise have been dropped due to unrelated callbacks before the
subscriber shows up.

Test: tests in statusbar/pipeline/mobile
Test: MobileConnectionTelephonySmokeTests
Fixes: 272784824
Change-Id: Id57cc4047d9290581e6b8eae813136ebfb8eb3be

packages/SystemUI/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryImpl.kt

@@ -18,7 +18,6 @@ package com.android.systemui.statusbar.pipeline.mobile.data.repository.prod
 
 import android.content.Context
 import android.content.IntentFilter
-import android.telephony.CellSignalStrength
 import android.telephony.CellSignalStrength.SIGNAL_STRENGTH_NONE_OR_UNKNOWN
 import android.telephony.CellSignalStrengthCdma
 import android.telephony.ServiceState
@@ -34,7 +33,6 @@ import android.telephony.TelephonyManager.EXTRA_SUBSCRIPTION_ID
 import android.telephony.TelephonyManager.NETWORK_TYPE_UNKNOWN
 import com.android.settingslib.Utils
 import com.android.systemui.broadcast.BroadcastDispatcher
-import com.android.systemui.common.coroutine.ConflatedCallbackFlow.conflatedCallbackFlow
 import com.android.systemui.dagger.qualifiers.Application
 import com.android.systemui.dagger.qualifiers.Background
 import com.android.systemui.log.table.TableLogBuffer
@@ -60,16 +58,14 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.asExecutor
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.mapLatest
 import kotlinx.coroutines.flow.mapNotNull
-import kotlinx.coroutines.flow.shareIn
+import kotlinx.coroutines.flow.scan
 import kotlinx.coroutines.flow.stateIn
 
 /**
@@ -101,8 +97,6 @@ class MobileConnectionRepositoryImpl(
         }
     }
 
-    private val telephonyCallbackEvent = MutableSharedFlow<Unit>(extraBufferCapacity = 1)
-
     /**
      * This flow defines the single shared connection to system_server via TelephonyCallback. Any
      * new callback should be added to this listener and funneled through callbackEvents via a data
@@ -110,9 +104,15 @@ class MobileConnectionRepositoryImpl(
      *
      * The reason we need to do this is because TelephonyManager limits the number of registered
      * listeners per-process, so we don't want to create a new listener for every callback.
+     *
+     * A note on the design for back pressure here: We use the [coalesce] operator here to change
+     * the backpressure strategy to store exactly the last callback event of _each type_ here, as
+     * opposed to the default strategy which is to drop the oldest event (regardless of type). This
+     * means that we should never miss any single event as long as the flow has been started.
      */
-    private val callbackEvents: SharedFlow<CallbackEvent> =
-        conflatedCallbackFlow {
+    private val callbackEvents: StateFlow<TelephonyCallbackState> = run {
+        val initial = TelephonyCallbackState()
+        callbackFlow {
                 val callback =
                     object :
                         TelephonyCallback(),
@@ -166,48 +166,50 @@ class MobileConnectionRepositoryImpl(
                 telephonyManager.registerTelephonyCallback(bgDispatcher.asExecutor(), callback)
                 awaitClose { telephonyManager.unregisterTelephonyCallback(callback) }
             }
-            .shareIn(scope, SharingStarted.WhileSubscribed())
+            .scan(initial = initial) { state, event -> state.applyEvent(event) }
+            .stateIn(scope = scope, started = SharingStarted.WhileSubscribed(), initial)
+    }
 
     override val isEmergencyOnly =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { it.serviceState.isEmergencyOnly }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val isRoaming =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { it.serviceState.roaming }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val operatorAlphaShort =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { it.serviceState.operatorAlphaShort }
             .stateIn(scope, SharingStarted.WhileSubscribed(), null)
 
     override val isInService =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { Utils.isInService(it.serviceState) }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val isGsm =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnSignalStrengthChanged>()
+            .mapNotNull { it.onSignalStrengthChanged }
             .map { it.signalStrength.isGsm }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val cdmaLevel =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnSignalStrengthChanged>()
+            .mapNotNull { it.onSignalStrengthChanged }
             .map {
                 it.signalStrength.getCellSignalStrengths(CellSignalStrengthCdma::class.java).let {
                     strengths ->
                     if (strengths.isNotEmpty()) {
                         strengths[0].level
                     } else {
-                        CellSignalStrength.SIGNAL_STRENGTH_NONE_OR_UNKNOWN
+                        SIGNAL_STRENGTH_NONE_OR_UNKNOWN
                     }
                 }
             }
@@ -215,19 +217,19 @@ class MobileConnectionRepositoryImpl(
 
     override val primaryLevel =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnSignalStrengthChanged>()
+            .mapNotNull { it.onSignalStrengthChanged }
             .map { it.signalStrength.level }
             .stateIn(scope, SharingStarted.WhileSubscribed(), SIGNAL_STRENGTH_NONE_OR_UNKNOWN)
 
     override val dataConnectionState =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnDataConnectionStateChanged>()
+            .mapNotNull { it.onDataConnectionStateChanged }
             .map { it.dataState.toDataConnectionType() }
             .stateIn(scope, SharingStarted.WhileSubscribed(), Disconnected)
 
     override val dataActivityDirection =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnDataActivity>()
+            .mapNotNull { it.onDataActivity }
             .map { it.direction.toMobileDataActivityModel() }
             .stateIn(
                 scope,
@@ -237,13 +239,13 @@ class MobileConnectionRepositoryImpl(
 
     override val carrierNetworkChangeActive =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnCarrierNetworkChange>()
+            .mapNotNull { it.onCarrierNetworkChange }
             .map { it.active }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val resolvedNetworkType =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnDisplayInfoChanged>()
+            .mapNotNull { it.onDisplayInfoChanged }
             .map {
                 if (it.telephonyDisplayInfo.overrideNetworkType != OVERRIDE_NETWORK_TYPE_NONE) {
                     OverrideNetworkType(
@@ -302,7 +304,8 @@ class MobileConnectionRepositoryImpl(
     override val dataEnabled = run {
         val initial = telephonyManager.isDataConnectionAllowed
         callbackEvents
-            .mapNotNull { (it as? CallbackEvent.OnDataEnabledChanged)?.enabled }
+            .mapNotNull { it.onDataEnabledChanged }
+            .map { it.enabled }
             .stateIn(scope, SharingStarted.WhileSubscribed(), initial)
     }
 
@@ -346,12 +349,41 @@ class MobileConnectionRepositoryImpl(
  * Wrap every [TelephonyCallback] we care about in a data class so we can accept them in a single
  * shared flow and then split them back out into other flows.
  */
-private sealed interface CallbackEvent {
+sealed interface CallbackEvent {
+    data class OnCarrierNetworkChange(val active: Boolean) : CallbackEvent
+    data class OnDataActivity(val direction: Int) : CallbackEvent
+    data class OnDataConnectionStateChanged(val dataState: Int) : CallbackEvent
+    data class OnDataEnabledChanged(val enabled: Boolean) : CallbackEvent
+    data class OnDisplayInfoChanged(val telephonyDisplayInfo: TelephonyDisplayInfo) : CallbackEvent
     data class OnServiceStateChanged(val serviceState: ServiceState) : CallbackEvent
     data class OnSignalStrengthChanged(val signalStrength: SignalStrength) : CallbackEvent
-    data class OnDataConnectionStateChanged(val dataState: Int) : CallbackEvent
-    data class OnDataActivity(val direction: Int) : CallbackEvent
-    data class OnCarrierNetworkChange(val active: Boolean) : CallbackEvent
-    data class OnDisplayInfoChanged(val telephonyDisplayInfo: TelephonyDisplayInfo) : CallbackEvent
-    data class OnDataEnabledChanged(val enabled: Boolean) : CallbackEvent
+}
+
+/**
+ * A simple box type for 1-to-1 mapping of [CallbackEvent] to the batched event. Used in conjunction
+ * with [scan] to make sure we don't drop important callbacks due to late subscribers
+ */
+data class TelephonyCallbackState(
+    val onDataActivity: CallbackEvent.OnDataActivity? = null,
+    val onCarrierNetworkChange: CallbackEvent.OnCarrierNetworkChange? = null,
+    val onDataConnectionStateChanged: CallbackEvent.OnDataConnectionStateChanged? = null,
+    val onDataEnabledChanged: CallbackEvent.OnDataEnabledChanged? = null,
+    val onDisplayInfoChanged: CallbackEvent.OnDisplayInfoChanged? = null,
+    val onServiceStateChanged: CallbackEvent.OnServiceStateChanged? = null,
+    val onSignalStrengthChanged: CallbackEvent.OnSignalStrengthChanged? = null,
+) {
+    fun applyEvent(event: CallbackEvent): TelephonyCallbackState {
+        return when (event) {
+            is CallbackEvent.OnCarrierNetworkChange -> copy(onCarrierNetworkChange = event)
+            is CallbackEvent.OnDataActivity -> copy(onDataActivity = event)
+            is CallbackEvent.OnDataConnectionStateChanged ->
+                copy(onDataConnectionStateChanged = event)
+            is CallbackEvent.OnDataEnabledChanged -> copy(onDataEnabledChanged = event)
+            is CallbackEvent.OnDisplayInfoChanged -> copy(onDisplayInfoChanged = event)
+            is CallbackEvent.OnServiceStateChanged -> {
+                copy(onServiceStateChanged = event)
+            }
+            is CallbackEvent.OnSignalStrengthChanged -> copy(onSignalStrengthChanged = event)
+        }
+    }
 }

packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryTest.kt

@@ -18,12 +18,10 @@ package com.android.systemui.statusbar.pipeline.mobile.data.repository.prod
 
 import android.content.Intent
 import android.telephony.CarrierConfigManager.KEY_INFLATE_SIGNAL_STRENGTH_BOOL
-import android.telephony.CellSignalStrengthCdma
 import android.telephony.NetworkRegistrationInfo
 import android.telephony.ServiceState
 import android.telephony.ServiceState.STATE_IN_SERVICE
 import android.telephony.ServiceState.STATE_OUT_OF_SERVICE
-import android.telephony.SignalStrength
 import android.telephony.TelephonyCallback
 import android.telephony.TelephonyCallback.DataActivityListener
 import android.telephony.TelephonyCallback.ServiceStateListener
@@ -69,6 +67,7 @@ import com.android.systemui.statusbar.pipeline.mobile.data.model.SystemUiCarrier
 import com.android.systemui.statusbar.pipeline.mobile.data.model.toNetworkNameModel
 import com.android.systemui.statusbar.pipeline.mobile.data.repository.FakeMobileConnectionsRepository
 import com.android.systemui.statusbar.pipeline.mobile.data.repository.MobileConnectionRepository.Companion.DEFAULT_NUM_LEVELS
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.prod.MobileTelephonyHelpers.signalStrength
 import com.android.systemui.statusbar.pipeline.mobile.util.FakeMobileMappingsProxy
 import com.android.systemui.statusbar.pipeline.shared.data.model.DataActivityModel
 import com.android.systemui.statusbar.pipeline.shared.data.model.toMobileDataActivityModel
@@ -76,7 +75,6 @@ import com.android.systemui.util.mockito.any
 import com.android.systemui.util.mockito.mock
 import com.android.systemui.util.mockito.whenever
 import com.google.common.truth.Truth.assertThat
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
@@ -126,7 +124,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
                 systemUiCarrierConfig,
                 fakeBroadcastDispatcher,
                 mobileMappings,
-                IMMEDIATE,
+                testDispatcher,
                 logger,
                 tableLogger,
                 testScope.backgroundScope,
@@ -156,13 +154,10 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
             val job = underTest.isEmergencyOnly.onEach { latest = it }.launchIn(this)
 
             val callback = getTelephonyCallbackForType<ServiceStateListener>()
-            val serviceState = ServiceState()
-            serviceState.isEmergencyOnly = true
-            callback.onServiceStateChanged(serviceState)
+            callback.onServiceStateChanged(ServiceState().also { it.isEmergencyOnly = true })
             assertThat(latest).isTrue()
 
-            serviceState.isEmergencyOnly = false
-            callback.onServiceStateChanged(serviceState)
+            callback.onServiceStateChanged(ServiceState().also { it.isEmergencyOnly = false })
 
             assertThat(latest).isFalse()
 
@@ -520,18 +515,15 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
             val cb = getTelephonyCallbackForType<ServiceStateListener>()
 
-            val serviceState = ServiceState()
-            serviceState.roaming = false
-
-            // CDMA roaming is off, GSM roaming is off
+            // CDMA roaming is off, GSM roaming is on
             whenever(telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber).thenReturn(ERI_OFF)
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = true })
 
             assertThat(latest).isFalse()
 
-            // CDMA roaming is off, GSM roaming is on
+            // CDMA roaming is on, GSM roaming is off
             whenever(telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber).thenReturn(ERI_ON)
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = false })
 
             assertThat(latest).isTrue()
 
@@ -568,20 +560,16 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
             var latest: Boolean? = null
             val job = underTest.isRoaming.onEach { latest = it }.launchIn(this)
 
-            val serviceState = ServiceState()
-            serviceState.roaming = false
-
             val cb = getTelephonyCallbackForType<ServiceStateListener>()
 
             // CDMA roaming is off, GSM roaming is off
             whenever(telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber).thenReturn(ERI_OFF)
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = false })
 
             assertThat(latest).isFalse()
 
             // CDMA roaming is off, GSM roaming is on
-            serviceState.roaming = true
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = true })
 
             assertThat(latest).isTrue()
 
@@ -733,20 +721,32 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
             var latest: Boolean? = null
             val job = underTest.isInService.onEach { latest = it }.launchIn(this)
 
-            val serviceState = ServiceState()
-            serviceState.voiceRegState = STATE_IN_SERVICE
-            serviceState.dataRegState = STATE_IN_SERVICE
-
-            getTelephonyCallbackForType<ServiceStateListener>().onServiceStateChanged(serviceState)
+            getTelephonyCallbackForType<ServiceStateListener>()
+                .onServiceStateChanged(
+                    ServiceState().also {
+                        it.voiceRegState = STATE_IN_SERVICE
+                        it.dataRegState = STATE_IN_SERVICE
+                    }
+                )
 
             assertThat(latest).isTrue()
 
-            serviceState.voiceRegState = STATE_OUT_OF_SERVICE
-            getTelephonyCallbackForType<ServiceStateListener>().onServiceStateChanged(serviceState)
+            getTelephonyCallbackForType<ServiceStateListener>()
+                .onServiceStateChanged(
+                    ServiceState().also {
+                        it.dataRegState = STATE_IN_SERVICE
+                        it.voiceRegState = STATE_OUT_OF_SERVICE
+                    }
+                )
             assertThat(latest).isTrue()
 
-            serviceState.dataRegState = STATE_OUT_OF_SERVICE
-            getTelephonyCallbackForType<ServiceStateListener>().onServiceStateChanged(serviceState)
+            getTelephonyCallbackForType<ServiceStateListener>()
+                .onServiceStateChanged(
+                    ServiceState().also {
+                        it.voiceRegState = STATE_OUT_OF_SERVICE
+                        it.dataRegState = STATE_OUT_OF_SERVICE
+                    }
+                )
             assertThat(latest).isFalse()
 
             job.cancel()
@@ -804,19 +804,6 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         return MobileTelephonyHelpers.getTelephonyCallbackForType(telephonyManager)
     }
 
-    /** Convenience constructor for SignalStrength */
-    private fun signalStrength(gsmLevel: Int, cdmaLevel: Int, isGsm: Boolean): SignalStrength {
-        val signalStrength = mock<SignalStrength>()
-        whenever(signalStrength.isGsm).thenReturn(isGsm)
-        whenever(signalStrength.level).thenReturn(gsmLevel)
-        val cdmaStrength =
-            mock<CellSignalStrengthCdma>().also { whenever(it.level).thenReturn(cdmaLevel) }
-        whenever(signalStrength.getCellSignalStrengths(CellSignalStrengthCdma::class.java))
-            .thenReturn(listOf(cdmaStrength))
-
-        return signalStrength
-    }
-
     private fun spnIntent(
         subId: Int = SUB_1_ID,
         showSpn: Boolean = true,
@@ -833,7 +820,6 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     companion object {
-        private val IMMEDIATE = Dispatchers.Main.immediate
         private const val SUB_1_ID = 1
 
         private val DEFAULT_NAME = NetworkNameModel.Default("default name")

packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionTelephonySmokeTests.kt

@@ -0,0 +1,341 @@
+/*
+ * Copyright (C) 2023 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.android.systemui.statusbar.pipeline.mobile.data.repository.prod
+
+import android.telephony.ServiceState
+import android.telephony.TelephonyCallback
+import android.telephony.TelephonyCallback.CarrierNetworkListener
+import android.telephony.TelephonyCallback.DataActivityListener
+import android.telephony.TelephonyCallback.DataConnectionStateListener
+import android.telephony.TelephonyCallback.DataEnabledListener
+import android.telephony.TelephonyCallback.DisplayInfoListener
+import android.telephony.TelephonyCallback.ServiceStateListener
+import android.telephony.TelephonyDisplayInfo
+import android.telephony.TelephonyManager
+import android.telephony.TelephonyManager.DATA_ACTIVITY_INOUT
+import android.telephony.TelephonyManager.NETWORK_TYPE_LTE
+import android.telephony.TelephonyManager.NETWORK_TYPE_UNKNOWN
+import androidx.test.filters.SmallTest
+import com.android.systemui.SysuiTestCase
+import com.android.systemui.log.table.TableLogBuffer
+import com.android.systemui.statusbar.pipeline.mobile.data.MobileInputLogger
+import com.android.systemui.statusbar.pipeline.mobile.data.model.DataConnectionState
+import com.android.systemui.statusbar.pipeline.mobile.data.model.NetworkNameModel
+import com.android.systemui.statusbar.pipeline.mobile.data.model.ResolvedNetworkType
+import com.android.systemui.statusbar.pipeline.mobile.data.model.SystemUiCarrierConfig
+import com.android.systemui.statusbar.pipeline.mobile.data.model.SystemUiCarrierConfigTest
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.FakeMobileConnectionsRepository
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.prod.MobileTelephonyHelpers.getTelephonyCallbackForType
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.prod.MobileTelephonyHelpers.signalStrength
+import com.android.systemui.statusbar.pipeline.mobile.util.FakeMobileMappingsProxy
+import com.android.systemui.statusbar.pipeline.shared.data.model.DataActivityModel
+import com.android.systemui.util.mockito.mock
+import com.android.systemui.util.mockito.whenever
+import com.google.common.truth.Truth.assertThat
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import org.junit.Before
+import org.junit.Test
+import org.mockito.Mock
+import org.mockito.MockitoAnnotations
+
+/**
+ * Test class to stress test the TelephonyCallbacks that we listen to. In particular, the callbacks
+ * all come back in on a single listener (for reasons defined in the system). This test is built to
+ * ensure that we don't miss any important callbacks.
+ *
+ * Kind of like an interaction test case build just for [TelephonyCallback]
+ *
+ * The list of telephony callbacks we use is: [TelephonyCallback.CarrierNetworkListener]
+ * [TelephonyCallback.DataActivityListener] [TelephonyCallback.DataConnectionStateListener]
+ * [TelephonyCallback.DataEnabledListener] [TelephonyCallback.DisplayInfoListener]
+ * [TelephonyCallback.ServiceStateListener] [TelephonyCallback.SignalStrengthsListener]
+ *
+ * Because each of these callbacks comes in on the same callbackFlow, collecting on a field backed
+ * by only a single callback can immediately create backpressure on the other fields related to a
+ * mobile connection.
+ *
+ * This test should be designed to test _at least_ each individual callback in a smoke-test fashion.
+ * The way we will achieve this is as follows:
+ * 1. Start up a listener (A) collecting on a field which is _not under test_
+ * 2. Send a single event to a telephony callback which supports the field under test (B)
+ * 3. Send many (may be as few as 2) events to the callback backing A to ensure we start seeing
+ *    backpressure on other fields NOTE: poor handling of backpressure here would normally cause B
+ *    to get dropped
+ * 4. Start up a new collector for B
+ * 5. Assert that B has the state sent in step #2
+ */
+@Suppress("EXPERIMENTAL_IS_NOT_ENABLED")
+@OptIn(ExperimentalCoroutinesApi::class)
+@SmallTest
+class MobileConnectionTelephonySmokeTests : SysuiTestCase() {
+    private lateinit var underTest: MobileConnectionRepositoryImpl
+    private lateinit var connectionsRepo: FakeMobileConnectionsRepository
+
+    @Mock private lateinit var telephonyManager: TelephonyManager
+    @Mock private lateinit var logger: MobileInputLogger
+    @Mock private lateinit var tableLogger: TableLogBuffer
+
+    private val mobileMappings = FakeMobileMappingsProxy()
+    private val systemUiCarrierConfig =
+        SystemUiCarrierConfig(
+            SUB_1_ID,
+            SystemUiCarrierConfigTest.createTestConfig(),
+        )
+
+    private val testDispatcher = UnconfinedTestDispatcher()
+    private val testScope = TestScope(testDispatcher)
+
+    @Before
+    fun setUp() {
+        MockitoAnnotations.initMocks(this)
+        whenever(telephonyManager.subscriptionId).thenReturn(SUB_1_ID)
+
+        connectionsRepo = FakeMobileConnectionsRepository(mobileMappings, tableLogger)
+
+        underTest =
+            MobileConnectionRepositoryImpl(
+                context,
+                SUB_1_ID,
+                DEFAULT_NAME,
+                SEP,
+                telephonyManager,
+                systemUiCarrierConfig,
+                fakeBroadcastDispatcher,
+                mobileMappings,
+                testDispatcher,
+                logger,
+                tableLogger,
+                testScope.backgroundScope,
+            )
+    }
+
+    @Test
+    fun carrierNetworkChangeListener_noisyActivity() =
+        testScope.runTest {
+            var latest: Boolean? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            val callback = getTelephonyCallbackForType<CarrierNetworkListener>()
+            callback.onCarrierNetworkChange(true)
+
+            flipActivity(100, activityCallback)
+
+            val job = underTest.carrierNetworkChangeActive.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isTrue()
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun dataActivityLate_noisyDisplayInfo() =
+        testScope.runTest {
+            var latest: DataActivityModel? = null
+
+            // start collecting displayInfo; don't care about the result
+            val displayInfoJob = underTest.resolvedNetworkType.launchIn(this)
+
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+            activityCallback.onDataActivity(DATA_ACTIVITY_INOUT)
+
+            val displayInfoCallback = getTelephonyCallbackForType<DisplayInfoListener>()
+            val type1 = NETWORK_TYPE_UNKNOWN
+            val type2 = NETWORK_TYPE_LTE
+            val t1 =
+                mock<TelephonyDisplayInfo>().also { whenever(it.networkType).thenReturn(type1) }
+            val t2 =
+                mock<TelephonyDisplayInfo>().also { whenever(it.networkType).thenReturn(type2) }
+
+            flipDisplayInfo(100, listOf(t1, t2), displayInfoCallback)
+
+            val job = underTest.dataActivityDirection.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest)
+                .isEqualTo(
+                    DataActivityModel(
+                        hasActivityIn = true,
+                        hasActivityOut = true,
+                    )
+                )
+
+            displayInfoJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun dataConnectionStateListener_noisyActivity() =
+        testScope.runTest {
+            var latest: DataConnectionState? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val connectionCallback = getTelephonyCallbackForType<DataConnectionStateListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            connectionCallback.onDataConnectionStateChanged(
+                TelephonyManager.DATA_CONNECTED,
+                200 /* unused */
+            )
+
+            // Send a bunch of events that we don't care about, to overrun the replay buffer
+            flipActivity(100, activityCallback)
+
+            val connectionJob = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isEqualTo(DataConnectionState.Connected)
+
+            activityJob.cancel()
+            connectionJob.cancel()
+        }
+
+    @Test
+    fun dataEnabledLate_noisyActivity() =
+        testScope.runTest {
+            var latest: Boolean? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val enabledCallback = getTelephonyCallbackForType<DataEnabledListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            enabledCallback.onDataEnabledChanged(true, 1 /* unused */)
+
+            // Send a bunch of events that we don't care about, to overrun the replay buffer
+            flipActivity(100, activityCallback)
+
+            val job = underTest.dataEnabled.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isTrue()
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun displayInfoLate_noisyActivity() =
+        testScope.runTest {
+            var latest: ResolvedNetworkType? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val displayInfoCallback = getTelephonyCallbackForType<DisplayInfoListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            val type = NETWORK_TYPE_LTE
+            val expected = ResolvedNetworkType.DefaultNetworkType(mobileMappings.toIconKey(type))
+            val ti = mock<TelephonyDisplayInfo>().also { whenever(it.networkType).thenReturn(type) }
+            displayInfoCallback.onDisplayInfoChanged(ti)
+
+            // Send a bunch of events that we don't care about, to overrun the replay buffer
+            flipActivity(100, activityCallback)
+
+            val job = underTest.resolvedNetworkType.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isEqualTo(expected)
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun serviceStateListener_noisyActivity() =
+        testScope.runTest {
+            var latest: Boolean? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val serviceStateCallback = getTelephonyCallbackForType<ServiceStateListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            // isEmergencyOnly comes in
+            val serviceState = ServiceState()
+            serviceState.isEmergencyOnly = true
+            serviceStateCallback.onServiceStateChanged(serviceState)
+
+            flipActivity(100, activityCallback)
+
+            val job = underTest.isEmergencyOnly.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isTrue()
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun signalStrengthsListenerLate_noisyActivity() =
+        testScope.runTest {
+            var latest: Int? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            val callback = getTelephonyCallbackForType<TelephonyCallback.SignalStrengthsListener>()
+            val strength = signalStrength(gsmLevel = 1, cdmaLevel = 2, isGsm = true)
+            callback.onSignalStrengthsChanged(strength)
+
+            flipActivity(100, activityCallback)
+
+            val job = underTest.cdmaLevel.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isEqualTo(2)
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    private fun flipActivity(
+        times: Int,
+        callback: DataActivityListener,
+    ) {
+        repeat(times) { index -> callback.onDataActivity(index % 4) }
+    }
+
+    private fun flipDisplayInfo(
+        times: Int,
+        infos: List<TelephonyDisplayInfo>,
+        callback: DisplayInfoListener,
+    ) {
+        val len = infos.size
+        repeat(times) { index -> callback.onDisplayInfoChanged(infos[index % len]) }
+    }
+
+    private inline fun <reified T> getTelephonyCallbackForType(): T {
+        return getTelephonyCallbackForType(telephonyManager)
+    }
+
+    companion object {
+        private const val SUB_1_ID = 1
+
+        private val DEFAULT_NAME = NetworkNameModel.Default("default name")
+        private const val SEP = "-"
+    }
+}

packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileTelephonyHelpers.kt

@@ -16,10 +16,14 @@
 
 package com.android.systemui.statusbar.pipeline.mobile.data.repository.prod
 
+import android.telephony.CellSignalStrengthCdma
+import android.telephony.SignalStrength
 import android.telephony.TelephonyCallback
 import android.telephony.TelephonyManager
 import com.android.systemui.util.mockito.any
 import com.android.systemui.util.mockito.argumentCaptor
+import com.android.systemui.util.mockito.mock
+import com.android.systemui.util.mockito.whenever
 import com.google.common.truth.Truth.assertThat
 import org.mockito.Mockito.verify
 
@@ -31,6 +35,19 @@ object MobileTelephonyHelpers {
         return callbackCaptor.allValues
     }
 
+    /** Convenience constructor for SignalStrength */
+    fun signalStrength(gsmLevel: Int, cdmaLevel: Int, isGsm: Boolean): SignalStrength {
+        val signalStrength = mock<SignalStrength>()
+        whenever(signalStrength.isGsm).thenReturn(isGsm)
+        whenever(signalStrength.level).thenReturn(gsmLevel)
+        val cdmaStrength =
+            mock<CellSignalStrengthCdma>().also { whenever(it.level).thenReturn(cdmaLevel) }
+        whenever(signalStrength.getCellSignalStrengths(CellSignalStrengthCdma::class.java))
+            .thenReturn(listOf(cdmaStrength))
+
+        return signalStrength
+    }
+
     inline fun <reified T> getTelephonyCallbackForType(mockTelephonyManager: TelephonyManager): T {
         val cbs = getTelephonyCallbacks(mockTelephonyManager).filterIsInstance<T>()
         assertThat(cbs.size).isEqualTo(1)