Add @Broadcast Handler, broadcastFlow() -> BroadcastSubscriber

Provides a background handler (thread) for Broadcast delivery.

Currently, broadcasts are flowed on a @Background CoroutineDispatcher
but before that, they are delivered to the app via the Main thread
Handler. This means broadcasts being processed during startup will
get deferred until Main thread idle (generally, right after onResume)

This change improves on broadcastFlow, providing a broadcast flow
factory (BroadcastSubscriber), which registers Receivers with a
singleton @Broadcast Handler. This makes broadcast delivery much
more reliable and predictable.

Test: manually. Observe broadcasts received while the process is frozen
      are now handled immediately after the unfreeze, and no longer
      blocked until after onResume.
Bug: 330561320
Flag: ACONFIG intentresolver/com.android.intentresolver.enable_private_profile
Change-Id: I8ecf151d3bf7627d9e3917fb9fecd78c1e201521

5 files changed, 134 insertions, 76 deletions

diff --git a/java/src/com/android/intentresolver/inject/ConcurrencyModule.kt b/java/src/com/android/intentresolver/inject/ConcurrencyModule.kt
index e0f8e88b..5fbdf090 100644
--- a/java/src/com/android/intentresolver/inject/ConcurrencyModule.kt
+++ b/java/src/com/android/intentresolver/inject/ConcurrencyModule.kt
@@ -16,6 +16,10 @@
 
 package com.android.intentresolver.inject
 
+import android.os.Handler
+import android.os.HandlerThread
+import android.os.Looper
+import android.os.Process
 import dagger.Module
 import dagger.Provides
 import dagger.hilt.InstallIn
@@ -26,6 +30,10 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 
+// thread
+private const val BROADCAST_SLOW_DISPATCH_THRESHOLD = 1000L
+private const val BROADCAST_SLOW_DELIVERY_THRESHOLD = 1000L
+
 @Module
 @InstallIn(SingletonComponent::class)
 object ConcurrencyModule {
@@ -40,4 +48,25 @@ object ConcurrencyModule {
         CoroutineScope(SupervisorJob() + mainDispatcher)
 
     @Provides @Background fun backgroundDispatcher(): CoroutineDispatcher = Dispatchers.IO
+
+    @Provides
+    @Singleton
+    @Broadcast
+    fun provideBroadcastLooper(): Looper {
+        val thread = HandlerThread("BroadcastReceiver", Process.THREAD_PRIORITY_BACKGROUND)
+        thread.start()
+        thread.looper.setSlowLogThresholdMs(
+            BROADCAST_SLOW_DISPATCH_THRESHOLD,
+            BROADCAST_SLOW_DELIVERY_THRESHOLD
+        )
+        return thread.looper
+    }
+
+    /** Provide a BroadcastReceiver Executor (for sending and receiving broadcasts). */
+    @Provides
+    @Singleton
+    @Broadcast
+    fun provideBroadcastHandler(@Broadcast looper: Looper): Handler {
+        return Handler(looper)
+    }
 }
diff --git a/java/src/com/android/intentresolver/inject/Qualifiers.kt b/java/src/com/android/intentresolver/inject/Qualifiers.kt
index f267328b..77315cac 100644
--- a/java/src/com/android/intentresolver/inject/Qualifiers.kt
+++ b/java/src/com/android/intentresolver/inject/Qualifiers.kt
@@ -35,6 +35,8 @@ annotation class ApplicationOwned
 @Retention(AnnotationRetention.RUNTIME)
 annotation class ApplicationUser
 
+@Qualifier @MustBeDocumented @Retention(AnnotationRetention.RUNTIME) annotation class Broadcast
+
 @Qualifier @MustBeDocumented @Retention(AnnotationRetention.RUNTIME) annotation class ProfileParent
 
 @Qualifier @MustBeDocumented @Retention(AnnotationRetention.RUNTIME) annotation class Background
diff --git a/java/src/com/android/intentresolver/v2/data/BroadcastFlow.kt b/java/src/com/android/intentresolver/v2/data/BroadcastFlow.kt
deleted file mode 100644
index 1a58afcb..00000000
--- a/java/src/com/android/intentresolver/v2/data/BroadcastFlow.kt
+++ /dev/null
@@ -1,46 +0,0 @@
-package com.android.intentresolver.v2.data
-
-import android.content.BroadcastReceiver
-import android.content.Context
-import android.content.Intent
-import android.content.IntentFilter
-import android.os.UserHandle
-import android.util.Log
-import kotlinx.coroutines.channels.awaitClose
-import kotlinx.coroutines.channels.onFailure
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.callbackFlow
-
-private const val TAG = "BroadcastFlow"
-
-/**
- * Returns a [callbackFlow] that, when collected, registers a broadcast receiver and emits a new
- * value whenever broadcast matching _filter_ is received. The result value will be computed using
- * [transform] and emitted if non-null.
- */
-internal fun <T> broadcastFlow(
-    context: Context,
-    filter: IntentFilter,
-    user: UserHandle,
-    transform: (Intent) -> T?
-): Flow<T> = callbackFlow {
-    val receiver =
-        object : BroadcastReceiver() {
-            override fun onReceive(context: Context, intent: Intent) {
-                transform(intent)?.also { result ->
-                    trySend(result).onFailure { Log.e(TAG, "Failed to send $result", it) }
-                }
-                    ?: Log.w(TAG, "Ignored broadcast $intent")
-            }
-        }
-
-    context.registerReceiverAsUser(
-        receiver,
-        user,
-        IntentFilter(filter),
-        null,
-        null,
-        Context.RECEIVER_NOT_EXPORTED
-    )
-    awaitClose { context.unregisterReceiver(receiver) }
-}
diff --git a/java/src/com/android/intentresolver/v2/data/BroadcastSubscriber.kt b/java/src/com/android/intentresolver/v2/data/BroadcastSubscriber.kt
new file mode 100644
index 00000000..f3013246
--- /dev/null
+++ b/java/src/com/android/intentresolver/v2/data/BroadcastSubscriber.kt
@@ -0,0 +1,73 @@
+/*
+ * Copyright (C) 2024 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.intentresolver.v2.data
+
+import android.content.BroadcastReceiver
+import android.content.Context
+import android.content.Intent
+import android.content.IntentFilter
+import android.os.Handler
+import android.os.UserHandle
+import android.util.Log
+import com.android.intentresolver.inject.Broadcast
+import dagger.hilt.android.qualifiers.ApplicationContext
+import javax.inject.Inject
+import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.channels.onFailure
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
+
+private const val TAG = "BroadcastSubscriber"
+
+class BroadcastSubscriber
+@Inject
+constructor(
+    @ApplicationContext private val context: Context,
+    @Broadcast private val handler: Handler
+) {
+    /**
+     * Returns a [callbackFlow] that, when collected, registers a broadcast receiver and emits a new
+     * value whenever broadcast matching _filter_ is received. The result value will be computed
+     * using [transform] and emitted if non-null.
+     */
+    fun <T> createFlow(
+        filter: IntentFilter,
+        user: UserHandle,
+        transform: (Intent) -> T?,
+    ): Flow<T> = callbackFlow {
+        val receiver =
+            object : BroadcastReceiver() {
+                override fun onReceive(context: Context, intent: Intent) {
+                    transform(intent)?.also { result ->
+                        trySend(result).onFailure { Log.e(TAG, "Failed to send $result", it) }
+                    }
+                        ?: Log.w(TAG, "Ignored broadcast $intent")
+                }
+            }
+
+        @Suppress("MissingPermission")
+        context.registerReceiverAsUser(
+            receiver,
+            user,
+            IntentFilter(filter),
+            null,
+            handler,
+            Context.RECEIVER_NOT_EXPORTED
+        )
+        awaitClose { context.unregisterReceiver(receiver) }
+    }
+}
diff --git a/java/src/com/android/intentresolver/v2/data/repository/UserRepository.kt b/java/src/com/android/intentresolver/v2/data/repository/UserRepository.kt
index 40672249..d196d3e6 100644
--- a/java/src/com/android/intentresolver/v2/data/repository/UserRepository.kt
+++ b/java/src/com/android/intentresolver/v2/data/repository/UserRepository.kt
@@ -16,7 +16,6 @@
 
 package com.android.intentresolver.v2.data.repository
 
-import android.content.Context
 import android.content.Intent
 import android.content.Intent.ACTION_MANAGED_PROFILE_AVAILABLE
 import android.content.Intent.ACTION_MANAGED_PROFILE_UNAVAILABLE
@@ -36,9 +35,8 @@ import androidx.annotation.VisibleForTesting
 import com.android.intentresolver.inject.Background
 import com.android.intentresolver.inject.Main
 import com.android.intentresolver.inject.ProfileParent
-import com.android.intentresolver.v2.data.broadcastFlow
+import com.android.intentresolver.v2.data.BroadcastSubscriber
 import com.android.intentresolver.v2.shared.model.User
-import dagger.hilt.android.qualifiers.ApplicationContext
 import javax.inject.Inject
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
@@ -88,6 +86,23 @@ internal data class UserWithState(val user: User, val available: Boolean)
 
 internal typealias UserStates = List<UserWithState>
 
+internal val userBroadcastActions =
+    setOf(
+        ACTION_PROFILE_ADDED,
+        ACTION_PROFILE_REMOVED,
+
+        // Quiet mode enabled/disabled for managed
+        // From: UserController.broadcastProfileAvailabilityChanges
+        // In response to setQuietModeEnabled
+        ACTION_MANAGED_PROFILE_AVAILABLE, // quiet mode, sent for manage profiles only
+        ACTION_MANAGED_PROFILE_UNAVAILABLE, // quiet mode, sent for manage profiles only
+
+        // Quiet mode toggled for profile type, requires flag 'android.os.allow_private_profile
+        // true'
+        ACTION_PROFILE_AVAILABLE, // quiet mode,
+        ACTION_PROFILE_UNAVAILABLE, // quiet mode, sent for any profile type
+    )
+
 /** Tracks and publishes state for the parent user and associated profiles. */
 class UserRepositoryImpl
 @VisibleForTesting
@@ -97,21 +112,26 @@ constructor(
     /** A flow of events which represent user-state changes from [UserManager]. */
     private val userEvents: Flow<UserEvent>,
     scope: CoroutineScope,
-    private val backgroundDispatcher: CoroutineDispatcher
+    private val backgroundDispatcher: CoroutineDispatcher,
 ) : UserRepository {
     @Inject
     constructor(
-        @ApplicationContext context: Context,
         @ProfileParent profileParent: UserHandle,
         userManager: UserManager,
         @Main scope: CoroutineScope,
-        @Background background: CoroutineDispatcher
+        @Background background: CoroutineDispatcher,
+        broadcastSubscriber: BroadcastSubscriber,
     ) : this(
         profileParent,
         userManager,
-        userEvents = userBroadcastFlow(context, profileParent),
+        userEvents =
+            broadcastSubscriber.createFlow(
+                createFilter(userBroadcastActions),
+                profileParent,
+                Intent::toUserEvent
+            ),
         scope,
-        background
+        background,
     )
 
     private fun debugLog(msg: () -> String) {
@@ -264,7 +284,7 @@ data class UnknownEvent(
 ) : UserEvent
 
 /** Used with [broadcastFlow] to transform a UserManager broadcast action into a [UserEvent]. */
-private fun Intent.toUserEvent(): UserEvent {
+internal fun Intent.toUserEvent(): UserEvent {
     val action = action
     val user = extras?.getParcelable(EXTRA_USER, UserHandle::class.java)
     val quietMode = extras?.getBoolean(EXTRA_QUIET_MODE, false)
@@ -280,30 +300,10 @@ private fun Intent.toUserEvent(): UserEvent {
     }
 }
 
-private fun createFilter(actions: Iterable<String>): IntentFilter {
+internal fun createFilter(actions: Iterable<String>): IntentFilter {
     return IntentFilter().apply { actions.forEach(::addAction) }
 }
 
 internal fun UserInfo?.isAvailable(): Boolean {
     return this?.isQuietModeEnabled != true
 }
-
-internal fun userBroadcastFlow(context: Context, profileParent: UserHandle): Flow<UserEvent> {
-    val userActions =
-        setOf(
-            ACTION_PROFILE_ADDED,
-            ACTION_PROFILE_REMOVED,
-
-            // Quiet mode enabled/disabled for managed
-            // From: UserController.broadcastProfileAvailabilityChanges
-            // In response to setQuietModeEnabled
-            ACTION_MANAGED_PROFILE_AVAILABLE, // quiet mode, sent for manage profiles only
-            ACTION_MANAGED_PROFILE_UNAVAILABLE, // quiet mode, sent for manage profiles only
-
-            // Quiet mode toggled for profile type, requires flag 'android.os.allow_private_profile
-            // true'
-            ACTION_PROFILE_AVAILABLE, // quiet mode,
-            ACTION_PROFILE_UNAVAILABLE, // quiet mode, sent for any profile type
-        )
-    return broadcastFlow(context, createFilter(userActions), profileParent, Intent::toUserEvent)
-}