[Status Bar] Use defaultDataSubId flow to trigger config re-fetch.

The root cause of b/285298304 is that, because no downstream work was
collecting on the `defaultDataSubId` flow *directly*, that flow was
never starting. So, we were never listening for
`ACTION_DEFAULT_DATA_SUBSCRIPTION_CHANGED` intent and were not
re-fetching the config when the default subscription changed.

However, this bug was masked by the `config_subIdChangeEvent_updated`
test, which was incorrectly passing. It was passing because
FakeBroadcastDispatcher was sending the
`ACTION_DEFAULT_DATA_SUBSCRIPTION_CHANGED` intent to the
`carrierConfigChangedEvent` receiver, even though that receiver did
*not* match the intent.

This CL:
1) Updates FakeBroadcastDispatcher to send intents only to receivers
   that match the intent.
2) Updates `config_subIdChangeEvent_updated` test to use the new
   FakeBroadcastDispatcher method, causing the test to fail.
3) Updates MobileConnectionsRepositoryImpl to collect on
   `defaultDataSubId` flow directly, causing the test to pass again.

Fixes: 285298304
Test: atest MobileConnectionsRepositoryTest#config_subIdChangeEvent_updated
Change-Id: I007751b4a6279569311362cd14410f053a6a47f2

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

@@ -57,7 +57,6 @@ 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.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.combine
@@ -205,9 +204,6 @@ constructor(
             }
             .stateIn(scope, SharingStarted.WhileSubscribed(), null)
 
-    private val defaultDataSubIdChangeEvent: MutableSharedFlow<Unit> =
-        MutableSharedFlow(extraBufferCapacity = 1)
-
     override val defaultDataSubId: StateFlow<Int> =
         broadcastDispatcher
             .broadcastFlow(
@@ -223,7 +219,6 @@ constructor(
                 initialValue = INVALID_SUBSCRIPTION_ID,
             )
             .onStart { emit(subscriptionManagerProxy.getDefaultDataSubscriptionId()) }
-            .onEach { defaultDataSubIdChangeEvent.tryEmit(Unit) }
             .stateIn(scope, SharingStarted.WhileSubscribed(), INVALID_SUBSCRIPTION_ID)
 
     private val carrierConfigChangedEvent =
@@ -232,7 +227,7 @@ constructor(
             .onEach { logger.logActionCarrierConfigChanged() }
 
     override val defaultDataSubRatConfig: StateFlow<Config> =
-        merge(defaultDataSubIdChangeEvent, carrierConfigChangedEvent)
+        merge(defaultDataSubId, carrierConfigChangedEvent)
             .onStart { emit(Unit) }
             .mapLatest { Config.readConfig(context) }
             .distinctUntilChanged()

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

@@ -1074,13 +1074,10 @@ class MobileConnectionsRepositoryTest : SysuiTestCase() {
             assertThat(configFromContext.showAtLeast3G).isTrue()
 
             // WHEN the change event is fired
-            fakeBroadcastDispatcher.registeredReceivers.forEach { receiver ->
-                receiver.onReceive(
-                    context,
-                    Intent(TelephonyManager.ACTION_DEFAULT_DATA_SUBSCRIPTION_CHANGED)
-                        .putExtra(PhoneConstants.SUBSCRIPTION_KEY, SUB_1_ID)
-                )
-            }
+            val intent =
+                Intent(TelephonyManager.ACTION_DEFAULT_DATA_SUBSCRIPTION_CHANGED)
+                    .putExtra(PhoneConstants.SUBSCRIPTION_KEY, SUB_1_ID)
+            fakeBroadcastDispatcher.sendIntentToMatchingReceiversOnly(context, intent)
 
             // THEN the config is updated
             assertTrue(latest!!.areEqual(configFromContext))

packages/SystemUI/tests/utils/src/com/android/systemui/broadcast/FakeBroadcastDispatcher.kt

@@ -18,6 +18,7 @@ package com.android.systemui.broadcast
 
 import android.content.BroadcastReceiver
 import android.content.Context
+import android.content.Intent
 import android.content.IntentFilter
 import android.os.Handler
 import android.os.Looper
@@ -31,6 +32,14 @@ import java.lang.IllegalStateException
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Executor
 
+/**
+ * A fake instance of [BroadcastDispatcher] for tests.
+ *
+ * Important: The *real* broadcast dispatcher will only send intents to receivers if the intent
+ * matches the [IntentFilter] that the [BroadcastReceiver] was registered with. This fake class does
+ * *not* do that matching by default. Use [sendIntentToMatchingReceiversOnly] to get the same
+ * matching behavior as the real broadcast dispatcher.
+ */
 class FakeBroadcastDispatcher(
     context: SysuiTestableContext,
     mainExecutor: Executor,
@@ -52,7 +61,10 @@ class FakeBroadcastDispatcher(
         PendingRemovalStore(logger)
     ) {
 
-    val registeredReceivers: MutableSet<BroadcastReceiver> = ConcurrentHashMap.newKeySet()
+    private val receivers: MutableSet<InternalReceiver> = ConcurrentHashMap.newKeySet()
+
+    val registeredReceivers: Set<BroadcastReceiver>
+        get() = receivers.map { it.receiver }.toSet()
 
     override fun registerReceiverWithHandler(
         receiver: BroadcastReceiver,
@@ -62,7 +74,7 @@ class FakeBroadcastDispatcher(
         @Context.RegisterReceiverFlags flags: Int,
         permission: String?
     ) {
-        registeredReceivers.add(receiver)
+        receivers.add(InternalReceiver(receiver, filter))
     }
 
     override fun registerReceiver(
@@ -73,15 +85,34 @@ class FakeBroadcastDispatcher(
         @Context.RegisterReceiverFlags flags: Int,
         permission: String?
     ) {
-        registeredReceivers.add(receiver)
+        receivers.add(InternalReceiver(receiver, filter))
     }
 
     override fun unregisterReceiver(receiver: BroadcastReceiver) {
-        registeredReceivers.remove(receiver)
+        receivers.removeIf { it.receiver == receiver }
     }
 
     override fun unregisterReceiverForUser(receiver: BroadcastReceiver, user: UserHandle) {
-        registeredReceivers.remove(receiver)
+        receivers.removeIf { it.receiver == receiver }
+    }
+
+    /**
+     * Sends the given [intent] to *only* the receivers that were registered with an [IntentFilter]
+     * that matches the intent.
+     */
+    fun sendIntentToMatchingReceiversOnly(context: Context, intent: Intent) {
+        receivers.forEach {
+            if (
+                it.filter.match(
+                    context.contentResolver,
+                    intent,
+                    /* resolve= */ false,
+                    /* logTag= */ "FakeBroadcastDispatcher",
+                ) > 0
+            ) {
+                it.receiver.onReceive(context, intent)
+            }
+        }
     }
 
     fun cleanUpReceivers(testName: String) {
@@ -91,6 +122,11 @@ class FakeBroadcastDispatcher(
                 throw IllegalStateException("Receiver not unregistered from dispatcher: $it")
             }
         }
-        registeredReceivers.clear()
+        receivers.clear()
     }
+
+    private data class InternalReceiver(
+        val receiver: BroadcastReceiver,
+        val filter: IntentFilter,
+    )
 }