Skip to content

Commit ffdf9a1

Browse files
MichalMatuMichalMatu
committed
fix: stabilize scanner result pipeline
1 parent 1ca48aa commit ffdf9a1

5 files changed

Lines changed: 97 additions & 36 deletions

File tree

‎core/data/src/main/java/io/blueeye/core/data/repository/handler/ble/BleScanHandler.kt‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import io.blueeye.core.model.EvidenceSource
2323
import io.blueeye.core.model.TrackingStatus
2424
import io.blueeye.core.scanner.analysis.BlePacketAnalyzer
2525
import io.blueeye.core.scanner.model.BleScanResultData
26+
import kotlinx.coroutines.CancellationException
2627
import kotlinx.coroutines.CoroutineScope
2728
import kotlinx.coroutines.Dispatchers
2829
import kotlinx.coroutines.SupervisorJob
@@ -95,8 +96,11 @@ class BleScanHandler @Inject constructor(
9596
recordPublicSafetyEvidenceEvents(ctx)
9697
queueAutoActiveProbe(ctx)
9798
}
99+
} catch (e: CancellationException) {
100+
throw e
98101
} catch (e: Exception) {
99102
android.util.Log.e(TAG, "Error handling scan result", e)
103+
throw e
100104
}
101105
}
102106

‎core/data/src/main/java/io/blueeye/core/permission/PermissionManager.kt‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ object PermissionManager {
1616
}
1717

1818
fun getMissingScannerStartupPermissions(context: Context): List<String> {
19-
return getMissingBlePermissions(context)
19+
return getMissingBlePermissions(context) + getMissingNotificationPermissions(context)
2020
}
2121

2222
fun missingPermissionsMessage(missingPermissions: List<String>): String {
@@ -42,6 +42,9 @@ object PermissionManager {
4242
} else {
4343
permissions += Manifest.permission.ACCESS_FINE_LOCATION
4444
}
45+
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.TIRAMISU) {
46+
permissions += Manifest.permission.POST_NOTIFICATIONS
47+
}
4548

4649
return permissions.distinct().toTypedArray()
4750
}
@@ -62,6 +65,15 @@ object PermissionManager {
6265
}
6366
}
6467

68+
private fun getMissingNotificationPermissions(context: Context): List<String> {
69+
if (Build.VERSION.SDK_INT < Build.VERSION_CODES.TIRAMISU) return emptyList()
70+
71+
return listOf(Manifest.permission.POST_NOTIFICATIONS)
72+
.filterNot { permission ->
73+
ContextCompat.checkSelfPermission(context, permission) == PackageManager.PERMISSION_GRANTED
74+
}
75+
}
76+
6577
private fun permissionLabel(permission: String): String {
6678
return when (permission) {
6779
Manifest.permission.BLUETOOTH_SCAN,

‎core/data/src/main/java/io/blueeye/core/scanner/manager/BleScanner.kt‎

Lines changed: 41 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,13 @@ import kotlinx.coroutines.CoroutineScope
1515
import kotlinx.coroutines.Dispatchers
1616
import kotlinx.coroutines.Job
1717
import kotlinx.coroutines.SupervisorJob
18+
import kotlinx.coroutines.channels.BufferOverflow
1819
import kotlinx.coroutines.channels.Channel
1920
import kotlinx.coroutines.delay
2021
import kotlinx.coroutines.flow.MutableStateFlow
2122
import kotlinx.coroutines.flow.asStateFlow
2223
import kotlinx.coroutines.launch
24+
import java.util.concurrent.atomic.AtomicLong
2325
import javax.inject.Inject
2426
import javax.inject.Singleton
2527

@@ -64,6 +66,8 @@ constructor(
6466
) {
6567
companion object {
6668
private const val TAG = "BleScanner"
69+
private const val SCAN_EVENT_CHANNEL_CAPACITY = 4_096
70+
private const val SCAN_EVENT_DROP_LOG_INTERVAL = 100L
6771
}
6872

6973
// Use SupervisorJob so child failures don't cancel the parent
@@ -74,10 +78,18 @@ constructor(
7478

7579
private var scanJob: Job? = null
7680
private var processingJob: Job? = null
81+
private val droppedScanEvents = AtomicLong(0L)
7782

7883
// Channel for sequential processing of scan results (replaces fire-and-forget)
79-
// BUFFERED channel to avoid blocking the BLE callback thread
80-
private val scanEventChannel = Channel<ScanEvent>(Channel.BUFFERED)
84+
// Explicit capacity avoids JVM's tiny default buffered channel under dense BLE traffic.
85+
private val scanEventChannel =
86+
Channel<ScanEvent>(
87+
capacity = SCAN_EVENT_CHANNEL_CAPACITY,
88+
onBufferOverflow = BufferOverflow.DROP_OLDEST,
89+
onUndeliveredElement = {
90+
recordDroppedScanEvent("queue overflow or shutdown")
91+
},
92+
)
8193

8294
init {
8395
// Start the event processing loop
@@ -152,7 +164,7 @@ constructor(
152164
val bleStarted = bleScanSource.start(
153165
macFilter = null,
154166
onResult = { result ->
155-
scanEventChannel.trySend(ScanEvent.BleResult(result))
167+
enqueueScanEvent(ScanEvent.BleResult(result))
156168
},
157169
onError = { errorCode ->
158170
val message = describeBleScanError(errorCode)
@@ -205,7 +217,7 @@ constructor(
205217
val bleStarted = bleScanSource.start(
206218
macFilter = macAddress,
207219
onResult = { result ->
208-
scanEventChannel.trySend(ScanEvent.BleResult(result))
220+
enqueueScanEvent(ScanEvent.BleResult(result))
209221
},
210222
onError = { errorCode ->
211223
val message = "Focused ${describeBleScanError(errorCode)}"
@@ -240,7 +252,7 @@ constructor(
240252
private fun startClassicDiscovery() {
241253
val classicStarted =
242254
classicScanSource.start { device, rssi, classOfDevice, uuids ->
243-
scanEventChannel.trySend(
255+
enqueueScanEvent(
244256
ScanEvent.ClassicResult(
245257
mac = device.address,
246258
name = device.name,
@@ -256,6 +268,20 @@ constructor(
256268
}
257269
}
258270

271+
private fun enqueueScanEvent(event: ScanEvent) {
272+
val enqueueResult = scanEventChannel.trySend(event)
273+
if (enqueueResult.isFailure) {
274+
recordDroppedScanEvent("enqueue failed")
275+
}
276+
}
277+
278+
private fun recordDroppedScanEvent(reason: String) {
279+
val dropped = droppedScanEvents.incrementAndGet()
280+
if (dropped == 1L || dropped % SCAN_EVENT_DROP_LOG_INTERVAL == 0L) {
281+
Log.w(TAG, "Dropped $dropped scan event(s): $reason")
282+
}
283+
}
284+
259285
/** Handles BLE scan result - called sequentially from event processor. */
260286
@SuppressLint("MissingPermission")
261287
private suspend fun handleBleResult(result: ScanResult) {
@@ -280,6 +306,11 @@ constructor(
280306
rawData = data.rawData,
281307
)
282308
repository.handleScanResult(params)
309+
.onFailure { error ->
310+
val message = "Scan processing failed: ${error.message ?: error.javaClass.simpleName}"
311+
Log.e(TAG, message, error)
312+
_state.value = ScannerState.Error(message)
313+
}
283314
}
284315

285316
private suspend fun handleClassicResult(result: ScanEvent.ClassicResult) {
@@ -289,7 +320,11 @@ constructor(
289320
rssi = result.rssi,
290321
classOfDevice = result.classOfDevice,
291322
serviceUuids = result.serviceUuids,
292-
)
323+
).onFailure { error ->
324+
val message = "Classic scan processing failed: ${error.message ?: error.javaClass.simpleName}"
325+
Log.e(TAG, message, error)
326+
_state.value = ScannerState.Error(message)
327+
}
293328
}
294329

295330
private fun describeBleScanError(errorCode: Int): String {

‎core/data/src/main/java/io/blueeye/service/ScannerService.kt‎

Lines changed: 38 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ class ScannerService : Service() {
6767
@Inject lateinit var deviceDao: io.blueeye.core.data.db.dao.DeviceDao
6868

6969
private val serviceScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
70+
private var startupJob: Job? = null
7071
private var cleanupJob: Job? = null
7172
private var scannerStateJob: Job? = null
7273
private var bluetoothStateReceiverRegistered = false
@@ -132,28 +133,38 @@ class ScannerService : Service() {
132133

133134
registerBluetoothStateReceiver()
134135

135-
// Rehydrate carryover tracker from database BEFORE starting scans
136-
serviceScope.launch {
137-
try {
138-
val existingDevices = deviceDao.getAllDevices()
139-
carryoverTracker.rehydrateFromDatabase(existingDevices)
140-
} catch (e: Exception) {
141-
Log.e("ScannerService", "Failed to rehydrate carryover tracker", e)
142-
}
143-
}
136+
startupJob =
137+
serviceScope.launch {
138+
try {
139+
val rehydrated =
140+
runCatching {
141+
val existingDevices = deviceDao.getAllDevices()
142+
carryoverTracker.rehydrateFromDatabase(existingDevices)
143+
}.onFailure { error ->
144+
Log.e("ScannerService", "Failed to rehydrate carryover tracker", error)
145+
}.isSuccess
146+
147+
if (!rehydrated) {
148+
failScanner("Scanner startup failed: tracking memory could not be restored.")
149+
return@launch
150+
}
144151

145-
// Reset Follow-Me tracking session state (session-based timing)
146-
bleScanHandler.resetSession()
152+
// Reset Follow-Me tracking session state (session-based timing)
153+
bleScanHandler.resetSession()
147154

148-
try {
149-
bleScanner.startScanning()
150-
observeScannerState()
151-
} catch (e: Exception) {
152-
failScanner("Scanner failed to start: ${e.message ?: e.javaClass.simpleName}", e)
153-
return
154-
}
155-
startCleanupJob()
156-
acquireWakeLock()
155+
try {
156+
bleScanner.startScanning()
157+
observeScannerState()
158+
} catch (e: Exception) {
159+
failScanner("Scanner failed to start: ${e.message ?: e.javaClass.simpleName}", e)
160+
return@launch
161+
}
162+
startCleanupJob()
163+
acquireWakeLock()
164+
} finally {
165+
startupJob = null
166+
}
167+
}
157168
}
158169

159170
private fun promoteToForeground(): Boolean {
@@ -175,6 +186,8 @@ class ScannerService : Service() {
175186
}
176187

177188
private fun stopForegroundService() {
189+
startupJob?.cancel()
190+
startupJob = null
178191
scannerStateJob?.cancel()
179192
scannerStateJob = null
180193
unregisterBluetoothStateReceiver()
@@ -217,6 +230,8 @@ class ScannerService : Service() {
217230

218231
override fun onDestroy() {
219232
super.onDestroy()
233+
startupJob?.cancel()
234+
startupJob = null
220235
scannerStateJob?.cancel()
221236
unregisterBluetoothStateReceiver()
222237
if (_scannerState.value !is ScannerRuntimeState.Error) {
@@ -271,6 +286,8 @@ class ScannerService : Service() {
271286
}
272287

273288
private fun cleanupAfterFailure() {
289+
startupJob?.cancel()
290+
startupJob = null
274291
scannerStateJob?.cancel()
275292
scannerStateJob = null
276293
cleanupJob?.cancel()
@@ -285,7 +302,7 @@ class ScannerService : Service() {
285302

286303
val filter = IntentFilter(BluetoothAdapter.ACTION_STATE_CHANGED)
287304
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.TIRAMISU) {
288-
registerReceiver(bluetoothStateReceiver, filter, Context.RECEIVER_NOT_EXPORTED)
305+
registerReceiver(bluetoothStateReceiver, filter, Context.RECEIVER_EXPORTED)
289306
} else {
290307
@Suppress("DEPRECATION")
291308
registerReceiver(bluetoothStateReceiver, filter)

‎feature/radar/src/main/java/io/blueeye/feature/radar/presentation/RadarViewModel.kt‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@ import io.blueeye.core.domain.usecase.GetScannedDevicesUseCase
1414
import io.blueeye.core.model.Device
1515
import io.blueeye.core.model.DeviceCalibrationLabel
1616
import io.blueeye.core.model.DeviceType
17-
import kotlinx.coroutines.FlowPreview
1817
import kotlinx.coroutines.flow.MutableStateFlow
1918
import kotlinx.coroutines.flow.SharingStarted
2019
import kotlinx.coroutines.flow.StateFlow
@@ -72,18 +71,12 @@ class RadarViewModel
7271
// Raw device stream (used for vendor extraction)
7372
private val rawDevicesFlow = getScannedDevicesUseCase(sinceSecondsAgo = 180)
7473

75-
// Live devices flow for UI
76-
private val devicesFlow =
77-
rawDevicesFlow
78-
.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), Result.success(emptyList()))
79-
8074
private val activeProbeFlow = deviceRepository.getActiveProbe()
8175

8276
// Główny strumień danych z filtrowaniem
83-
@OptIn(FlowPreview::class)
8477
val uiState: StateFlow<RadarUiState> =
8578
combine(
86-
devicesFlow,
79+
rawDevicesFlow,
8780
baselineDevices,
8881
_filter,
8982
activeProbeFlow,

0 commit comments

Comments
 (0)