Skip to content

Commit 8dbed15

Browse files
mushuiyu886mushuiyu886
committed
fix(android): queue notification events until node connect
1 parent 3137110 commit 8dbed15

5 files changed

Lines changed: 385 additions & 24 deletions

File tree

‎apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt‎

Lines changed: 72 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,9 +72,52 @@ import kotlinx.serialization.json.JsonArray
7272
import kotlinx.serialization.json.JsonObject
7373
import kotlinx.serialization.json.JsonPrimitive
7474
import kotlinx.serialization.json.buildJsonObject
75+
import java.util.ArrayDeque
7576
import java.util.UUID
7677
import java.util.concurrent.atomic.AtomicLong
7778

79+
private const val NOTIFICATIONS_CHANGED_EVENT = "notifications.changed"
80+
private const val MAX_PENDING_NOTIFICATION_EVENTS = 128
81+
82+
internal data class PendingNotificationEvent(
83+
val event: String,
84+
val payloadJson: String?,
85+
)
86+
87+
internal class PendingNotificationEventQueue(
88+
private val maxSize: Int = MAX_PENDING_NOTIFICATION_EVENTS,
89+
) {
90+
private val lock = Any()
91+
private val pending = ArrayDeque<PendingNotificationEvent>()
92+
93+
fun add(event: PendingNotificationEvent) {
94+
synchronized(lock) {
95+
while (pending.size >= maxSize) {
96+
pending.removeFirst()
97+
}
98+
pending.addLast(event)
99+
}
100+
}
101+
102+
fun drain(): List<PendingNotificationEvent> =
103+
synchronized(lock) {
104+
val drained = pending.toList()
105+
pending.clear()
106+
drained
107+
}
108+
109+
fun prepend(events: List<PendingNotificationEvent>) {
110+
synchronized(lock) {
111+
for (event in events.asReversed()) {
112+
while (pending.size >= maxSize) {
113+
pending.removeLast()
114+
}
115+
pending.addFirst(event)
116+
}
117+
}
118+
}
119+
}
120+
78121
/**
79122
* Process runtime that owns gateway sessions, node command handlers, capture managers, and UI-facing state.
80123
*/
@@ -421,6 +464,7 @@ class NodeRuntime(
421464
private val canvasRehydrateSeq = AtomicLong(0)
422465

423466
@Volatile private var nodePresenceAliveLastSuccessAtMs: Long? = null
467+
private val pendingNotificationEvents = PendingNotificationEventQueue()
424468
private var operatorConnected = false
425469
private var operatorStatusText: String = "Offline"
426470
private var nodeStatusText: String = "Offline"
@@ -501,6 +545,8 @@ class NodeRuntime(
501545
updateStatus()
502546
showLocalCanvasOnConnect()
503547
publishNodePresenceAliveBeacon(NodePresenceAliveBeacon.Trigger.Connect)
548+
flushQueuedNotificationEvents()
549+
DeviceNotificationListenerService.reconcileActiveNotifications()
504550
val endpoint = connectedEndpoint
505551
val auth = activeGatewayAuth
506552
if (endpoint != null && auth != null) {
@@ -530,7 +576,32 @@ class NodeRuntime(
530576
init {
531577
DeviceNotificationListenerService.setNodeEventSink { event, payloadJson ->
532578
scope.launch {
533-
nodeSession.sendNodeEvent(event = event, payloadJson = payloadJson)
579+
sendNotificationEventOrQueue(event = event, payloadJson = payloadJson)
580+
}
581+
}
582+
}
583+
584+
private suspend fun sendNotificationEventOrQueue(
585+
event: String,
586+
payloadJson: String?,
587+
) {
588+
val sent = nodeSession.sendNodeEvent(event = event, payloadJson = payloadJson)
589+
if (!sent && event == NOTIFICATIONS_CHANGED_EVENT) {
590+
pendingNotificationEvents.add(
591+
PendingNotificationEvent(event = event, payloadJson = payloadJson),
592+
)
593+
}
594+
}
595+
596+
private fun flushQueuedNotificationEvents() {
597+
scope.launch {
598+
val drained = pendingNotificationEvents.drain()
599+
for ((index, pending) in drained.withIndex()) {
600+
val sent = nodeSession.sendNodeEvent(event = pending.event, payloadJson = pending.payloadJson)
601+
if (!sent) {
602+
pendingNotificationEvents.prepend(drained.drop(index))
603+
return@launch
604+
}
534605
}
535606
}
536607
}

‎apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt‎

Lines changed: 44 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -199,6 +199,8 @@ class GatewaySession(
199199

200200
@Volatile private var currentConnection: Connection? = null
201201

202+
@Volatile private var connectingConnection: Connection? = null
203+
202204
// One reconnect can retry a shared-token mismatch by pairing the shared token with the stored device token.
203205
@Volatile private var pendingDeviceTokenRetry = false
204206

@@ -216,34 +218,34 @@ class GatewaySession(
216218
options: GatewayConnectOptions,
217219
tls: GatewayTlsParams? = null,
218220
) {
219-
val connectionToClose: Connection?
221+
val connectionsToClose: List<Connection>
220222
synchronized(lifecycleLock) {
221223
desired = DesiredConnection(endpoint, token, bootstrapToken, password, options, tls)
222224
pendingDeviceTokenRetry = false
223225
deviceTokenRetryBudgetUsed = false
224226
reconnectPausedForAuthFailure = false
225-
connectionToClose = currentConnection
227+
connectionsToClose = listOfNotNull(currentConnection, connectingConnection).distinct()
226228
if (job?.isActive != true) {
227229
job = scope.launch(Dispatchers.IO) { runLoop() }
228230
}
229231
}
230-
connectionToClose?.closeQuietly()
232+
connectionsToClose.forEach { it.closeQuietly() }
231233
}
232234

233235
/** Clears desired connection state, closes the socket, and stops reconnect attempts. */
234236
fun disconnect() {
235237
val jobToCancel: Job?
236-
val connectionToClose: Connection?
238+
val connectionsToClose: List<Connection>
237239
synchronized(lifecycleLock) {
238240
desired = null
239241
pendingDeviceTokenRetry = false
240242
deviceTokenRetryBudgetUsed = false
241243
reconnectPausedForAuthFailure = false
242-
connectionToClose = currentConnection
244+
connectionsToClose = listOfNotNull(currentConnection, connectingConnection).distinct()
243245
jobToCancel = job
244246
job = null
245247
}
246-
connectionToClose?.closeQuietly()
248+
connectionsToClose.forEach { it.closeQuietly() }
247249
scope.launch(Dispatchers.IO) {
248250
jobToCancel?.cancelAndJoin()
249251
if (desired == null) {
@@ -257,7 +259,7 @@ class GatewaySession(
257259
/** Forces the current socket closed so the loop reconnects to the current desired endpoint. */
258260
fun reconnect() {
259261
reconnectPausedForAuthFailure = false
260-
currentConnection?.closeQuietly()
262+
listOfNotNull(currentConnection, connectingConnection).distinct().forEach { it.closeQuietly() }
261263
}
262264

263265
fun currentCanvasHostUrl(): String? = pluginSurfaceUrls["canvas"]
@@ -409,7 +411,16 @@ class GatewaySession(
409411
val error: ErrorShape?,
410412
)
411413

414+
private fun promoteConnectedConnection(conn: Connection): Boolean =
415+
synchronized(lifecycleLock) {
416+
if (desired != conn.desiredTarget || connectingConnection !== conn) return@synchronized false
417+
currentConnection = conn
418+
connectingConnection = null
419+
true
420+
}
421+
412422
private inner class Connection(
423+
val desiredTarget: DesiredConnection,
413424
val endpoint: GatewayEndpoint,
414425
private val token: String?,
415426
private val bootstrapToken: String?,
@@ -663,6 +674,9 @@ class GatewaySession(
663674
}
664675
throw GatewayConnectFailure(error)
665676
}
677+
if (!promoteConnectedConnection(this)) {
678+
throw IllegalStateException("connection no longer desired")
679+
}
666680
handleConnectSuccess(res, identity.deviceId, selectedAuth.authSource)
667681
connectDeferred.complete(Unit)
668682
}
@@ -1093,18 +1107,34 @@ class GatewaySession(
10931107
withContext(Dispatchers.IO) {
10941108
val conn =
10951109
Connection(
1096-
target.endpoint,
1097-
target.token,
1098-
target.bootstrapToken,
1099-
target.password,
1100-
target.options,
1101-
target.tls,
1110+
desiredTarget = target,
1111+
endpoint = target.endpoint,
1112+
token = target.token,
1113+
bootstrapToken = target.bootstrapToken,
1114+
password = target.password,
1115+
options = target.options,
1116+
tls = target.tls,
11021117
)
1103-
currentConnection = conn
1118+
val shouldConnect =
1119+
synchronized(lifecycleLock) {
1120+
if (desired == target) {
1121+
connectingConnection = conn
1122+
true
1123+
} else {
1124+
false
1125+
}
1126+
}
1127+
if (!shouldConnect) {
1128+
conn.closeQuietly()
1129+
return@withContext
1130+
}
11041131
try {
11051132
conn.connect()
11061133
conn.awaitClose()
11071134
} finally {
1135+
if (connectingConnection === conn) {
1136+
connectingConnection = null
1137+
}
11081138
if (currentConnection === conn) {
11091139
currentConnection = null
11101140
pluginSurfaceUrls = emptyMap()

‎apps/android/app/src/main/java/ai/openclaw/app/node/DeviceNotificationListenerService.kt‎

Lines changed: 55 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -105,13 +105,17 @@ private object DeviceNotificationStore {
105105
private val lock = Any()
106106
private var connected = false
107107
private val byKey = LinkedHashMap<String, DeviceNotificationEntry>()
108+
private val postedEventKeys = LinkedHashSet<String>()
108109

109110
fun replace(entries: List<DeviceNotificationEntry>) {
110111
synchronized(lock) {
111112
byKey.clear()
113+
val activeKeys = LinkedHashSet<String>()
112114
for (entry in entries) {
113115
byKey[entry.key] = entry
116+
activeKeys += entry.key
114117
}
118+
postedEventKeys.retainAll(activeKeys)
115119
}
116120
}
117121

@@ -124,15 +128,30 @@ private object DeviceNotificationStore {
124128
fun remove(key: String) {
125129
synchronized(lock) {
126130
byKey.remove(key)
131+
postedEventKeys.remove(key)
127132
}
128133
}
129134

135+
fun notePostedEvent(key: String) {
136+
synchronized(lock) {
137+
postedEventKeys += key
138+
}
139+
}
140+
141+
fun entriesMissingPostedEvents(): List<DeviceNotificationEntry> =
142+
synchronized(lock) {
143+
byKey.values
144+
.filter { !postedEventKeys.contains(it.key) }
145+
.sortedByDescending { it.postTimeMs }
146+
}
147+
130148
fun setConnected(value: Boolean) {
131149
synchronized(lock) {
132150
connected = value
133151
if (!value) {
134152
// Android invalidates activeNotifications when the listener disconnects.
135153
byKey.clear()
154+
postedEventKeys.clear()
136155
}
137156
}
138157
}
@@ -150,6 +169,21 @@ private object DeviceNotificationStore {
150169
}
151170
}
152171

172+
internal fun replayMissedActiveNotificationEntries(
173+
entries: List<DeviceNotificationEntry>,
174+
emitPosted: (DeviceNotificationEntry) -> Boolean,
175+
): Int {
176+
DeviceNotificationStore.replace(entries)
177+
var replayed = 0
178+
for (entry in DeviceNotificationStore.entriesMissingPostedEvents()) {
179+
if (emitPosted(entry)) {
180+
DeviceNotificationStore.notePostedEvent(entry.key)
181+
replayed += 1
182+
}
183+
}
184+
return replayed
185+
}
186+
153187
/**
154188
* Android notification listener that mirrors notification state and executes gateway actions.
155189
*/
@@ -161,7 +195,7 @@ class DeviceNotificationListenerService : NotificationListenerService() {
161195
super.onListenerConnected()
162196
activeService = this
163197
DeviceNotificationStore.setConnected(true)
164-
refreshActiveNotifications()
198+
reconcileActiveNotifications()
165199
}
166200

167201
override fun onListenerDisconnected() {
@@ -187,8 +221,7 @@ class DeviceNotificationListenerService : NotificationListenerService() {
187221
if (entry.packageName == packageName) {
188222
return
189223
}
190-
val payload = notificationChangedPayload(entry) ?: return
191-
emitNotificationsChanged(payload)
224+
emitPostedNotification(entry)
192225
}
193226

194227
override fun onNotificationRemoved(sbn: StatusBarNotification?) {
@@ -217,6 +250,13 @@ class DeviceNotificationListenerService : NotificationListenerService() {
217250
emitNotificationsChanged(payload)
218251
}
219252

253+
private fun emitPostedNotification(entry: DeviceNotificationEntry): Boolean {
254+
val payload = notificationChangedPayload(entry) ?: return false
255+
if (!emitNotificationsChanged(payload)) return false
256+
DeviceNotificationStore.notePostedEvent(entry.key)
257+
return true
258+
}
259+
220260
private fun notificationChangedPayload(entry: DeviceNotificationEntry): String? =
221261
notificationChangedPayload(
222262
entry = entry,
@@ -272,14 +312,14 @@ class DeviceNotificationListenerService : NotificationListenerService() {
272312
}.toString()
273313
}
274314

275-
private fun refreshActiveNotifications() {
315+
private fun reconcileActiveNotifications() {
276316
val entries =
277317
runCatching {
278318
activeNotifications
279319
?.mapNotNull { it.toEntry() }
280320
?: emptyList()
281321
}.getOrElse { emptyList() }
282-
DeviceNotificationStore.replace(entries)
322+
replayMissedActiveNotificationEntries(entries, ::emitPostedNotification)
283323
}
284324

285325
private fun StatusBarNotification.toEntry(): DeviceNotificationEntry {
@@ -361,6 +401,11 @@ class DeviceNotificationListenerService : NotificationListenerService() {
361401
enabled: Boolean = isAccessEnabled(context),
362402
): DeviceNotificationSnapshot = DeviceNotificationStore.snapshot(enabled = enabled)
363403

404+
/** Replays active posted notifications that were not emitted before the gateway became ready. */
405+
fun reconcileActiveNotifications() {
406+
activeService?.reconcileActiveNotifications()
407+
}
408+
364409
/** Asks Android to rebind the listener after settings grant access but callbacks have not arrived. */
365410
fun requestServiceRebind(context: Context) {
366411
runCatching {
@@ -390,11 +435,12 @@ class DeviceNotificationListenerService : NotificationListenerService() {
390435
return service.executeActionInternal(request)
391436
}
392437

393-
private fun emitNotificationsChanged(payloadJson: String) {
438+
private fun emitNotificationsChanged(payloadJson: String): Boolean =
394439
runCatching {
395-
nodeEventSink?.invoke(NOTIFICATIONS_CHANGED_EVENT, payloadJson)
396-
}
397-
}
440+
val sink = nodeEventSink ?: return false
441+
sink.invoke(NOTIFICATIONS_CHANGED_EVENT, payloadJson)
442+
true
443+
}.getOrDefault(false)
398444

399445
private fun rememberRecentPackage(packageName: String?) {
400446
val service = activeService ?: return

0 commit comments

Comments
 (0)