permissionlesstech/bitchat-android
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat-android.git synced 2026-07-25 08:25:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

simplify

Browse Source
This commit is contained in:
callebtc
2025-10-21 13:25:39 +02:00
parent 6fb45f730a
commit 27c4d363b4
1 changed files with 40 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt
+40 -65
View File
@@ -114,85 +114,60 @@ class BluetoothPacketBroadcaster(
private val transferJobs = ConcurrentHashMap<String, Job>() private val transferJobs = ConcurrentHashMap<String, Job>()
// Replace simple actor FIFO with a priority-aware channel + processor // Replace simple actor FIFO with a priority-aware channel + processor
@Volatile private val broadcasterChannel = Channel<BroadcastRequest>(capacity = Channel.UNLIMITED)
private var broadcasterChannel = Channel<BroadcastRequest>(capacity = Channel.UNLIMITED)
private val sequenceCounter = AtomicLong(0) private val sequenceCounter = AtomicLong(0)
// Track processor state for recovery
@Volatile
private var processorJob: Job? = null
private val processorLock = Any()
init { init {
startPriorityProcessor() startPriorityProcessor()
} }
private fun startPriorityProcessor() { private fun startPriorityProcessor() {
synchronized(processorLock) { broadcasterScope.launch {
// Cancel existing processor if running Log.d(TAG, "🎭 Priority packet broadcaster processor started")
processorJob?.cancel()
// Create new channel for fresh start // Min-heap by (primaryPriority, secondaryPriority, sequence)
broadcasterChannel = Channel(capacity = Channel.UNLIMITED) val priorityQueue = PriorityQueue<QueuedBroadcast>(11) { a, b ->
when {
processorJob = broadcasterScope.launch { a.primaryPriority != b.primaryPriority -> a.primaryPriority - b.primaryPriority
var processorException: Exception? = null a.secondaryPriority != b.secondaryPriority -> a.secondaryPriority - b.secondaryPriority
Log.d(TAG, "🎭 Priority packet broadcaster processor started") else -> (a.sequence - b.sequence).coerceIn(Int.MIN_VALUE.toLong(), Int.MAX_VALUE.toLong()).toInt()
// Min-heap by (primaryPriority, secondaryPriority, sequence)
val priorityQueue = PriorityQueue<QueuedBroadcast>(11) { a, b ->
when {
a.primaryPriority != b.primaryPriority -> a.primaryPriority - b.primaryPriority
a.secondaryPriority != b.secondaryPriority -> a.secondaryPriority - b.secondaryPriority
else -> (a.sequence - b.sequence).coerceIn(Int.MIN_VALUE.toLong(), Int.MAX_VALUE.toLong()).toInt()
}
} }
}
try {
while (isActive) {
// If priority queue is empty, suspend to receive at least one item
if (priorityQueue.isEmpty()) {
val first = broadcasterChannel.receiveCatching().getOrNull() ?: break
priorityQueue.offer(computeQueuedBroadcast(first))
}
// Drain any immediately available items without suspending var processorException: Exception? = null
while (true) { try {
val received = broadcasterChannel.tryReceive().getOrNull() ?: break while (isActive) {
priorityQueue.offer(computeQueuedBroadcast(received)) // If priority queue is empty, suspend to receive at least one item
} if (priorityQueue.isEmpty()) {
val first = broadcasterChannel.receiveCatching().getOrNull() ?: break
priorityQueue.offer(computeQueuedBroadcast(first))
}
// Process one highest-priority item // Drain any immediately available items without suspending
val next = priorityQueue.poll() ?: continue while (true) {
try { val received = broadcasterChannel.tryReceive().getOrNull() ?: break
broadcastSinglePacketInternal(next.request.routed, next.request.gattServer, next.request.characteristic) priorityQueue.offer(computeQueuedBroadcast(received))
} catch (e: Exception) {
// Log the broadcast failure - this is likely a transient BLE error
Log.e(TAG, "❌ Broadcast failed, will attempt recovery: ${e.message}", e)
processorException = e
// Don't throw - instead break and trigger recovery
break
}
} }
} catch (e: Exception) {
Log.e(TAG, "❌ Priority processor loop terminated due to exception: ${e.message}", e) // Process one highest-priority item
processorException = e val next = priorityQueue.poll() ?: continue
} finally { try {
// Close the current channel broadcastSinglePacketInternal(next.request.routed, next.request.gattServer, next.request.characteristic)
broadcasterChannel.close(processorException) } catch (e: Exception) {
Log.w(TAG, "🎭 Priority packet broadcaster processor terminated, attempting recovery...") // BLE error during broadcast - log and save for channel closure
Log.e(TAG, "❌ Broadcast failed: ${e.message}", e)
// Schedule automatic recovery after a short delay processorException = e
if (processorException != null && broadcasterScope.isActive) { throw e // Re-throw to exit loop and close channel
broadcasterScope.launch {
delay(1000) // Wait 1 second before recovery
Log.d(TAG, "🔄 Attempting to restart priority processor...")
startPriorityProcessor()
}
} else {
Log.d(TAG, "🎭 Priority packet broadcaster processor shut down (no recovery scheduled)")
} }
} }
} catch (e: Exception) {
Log.e(TAG, "❌ Priority processor loop terminated: ${e.message}", e)
processorException = e
} finally {
// CRITICAL: Close channel so producers fail fast instead of accumulating packets
// This triggers the fallback path in broadcastSinglePacket()
broadcasterChannel.close(processorException)
Log.d(TAG, "🎭 Priority packet broadcaster processor terminated, channel closed")
} }
} }
} }