mirror of
https://github.com/permissionlesstech/bitchat-android.git
synced 2026-07-26 17:25:24 +00:00
263 lines
10 KiB
Kotlin
263 lines
10 KiB
Kotlin
package com.bitchat.android.mesh
|
|
|
|
import android.bluetooth.BluetoothDevice
|
|
import android.bluetooth.BluetoothGatt
|
|
import android.bluetooth.BluetoothGattCharacteristic
|
|
import android.bluetooth.BluetoothGattServer
|
|
import android.util.Log
|
|
import com.bitchat.android.protocol.SpecialRecipients
|
|
import com.bitchat.android.model.RoutedPacket
|
|
import kotlinx.coroutines.CoroutineScope
|
|
import kotlinx.coroutines.Dispatchers
|
|
import kotlinx.coroutines.SupervisorJob
|
|
import kotlinx.coroutines.cancel
|
|
import kotlinx.coroutines.delay
|
|
import kotlinx.coroutines.isActive
|
|
import kotlinx.coroutines.launch
|
|
import kotlinx.coroutines.channels.Channel
|
|
import kotlinx.coroutines.channels.actor
|
|
|
|
/**
|
|
* Handles packet broadcasting to connected devices using actor pattern for serialization
|
|
*
|
|
* SERIALIZATION FIX: Uses Kotlin coroutine actor to serialize all packet broadcasting
|
|
* This prevents race conditions when multiple threads try to broadcast simultaneously
|
|
*/
|
|
class BluetoothPacketBroadcaster(
|
|
private val connectionScope: CoroutineScope,
|
|
private val connectionTracker: BluetoothConnectionTracker,
|
|
private val fragmentManager: FragmentManager?
|
|
) {
|
|
|
|
companion object {
|
|
private const val TAG = "BluetoothPacketBroadcaster"
|
|
private const val CLEANUP_DELAY = 500L
|
|
}
|
|
|
|
// Data class to hold broadcast request information
|
|
private data class BroadcastRequest(
|
|
val routed: RoutedPacket,
|
|
val gattServer: BluetoothGattServer?,
|
|
val characteristic: BluetoothGattCharacteristic?
|
|
)
|
|
|
|
// Actor scope for the broadcaster
|
|
private val broadcasterScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
|
|
|
// SERIALIZATION: Actor to serialize all broadcast operations
|
|
@OptIn(kotlinx.coroutines.ObsoleteCoroutinesApi::class)
|
|
private val broadcasterActor = broadcasterScope.actor<BroadcastRequest>(
|
|
capacity = Channel.UNLIMITED
|
|
) {
|
|
Log.d(TAG, "🎭 Created packet broadcaster actor")
|
|
try {
|
|
for (request in channel) {
|
|
Log.d(TAG, "Processing broadcast for packet type ${request.routed.packet.type} (serialized)")
|
|
broadcastSinglePacketInternal(request.routed, request.gattServer, request.characteristic)
|
|
Log.d(TAG, "Completed broadcast for packet type ${request.routed.packet.type}")
|
|
}
|
|
} finally {
|
|
Log.d(TAG, "🎭 Packet broadcaster actor terminated")
|
|
}
|
|
}
|
|
|
|
fun broadcastPacket(
|
|
routed: RoutedPacket,
|
|
gattServer: BluetoothGattServer?,
|
|
characteristic: BluetoothGattCharacteristic?
|
|
) {
|
|
val packet = routed.packet
|
|
val data = packet.toBinaryData() ?: return
|
|
// Check if we need to fragment
|
|
if (fragmentManager != null) {
|
|
val fragments = fragmentManager.createFragments(packet)
|
|
if (fragments.size > 1) {
|
|
Log.d(TAG, "Fragmenting packet into ${fragments.size} fragments")
|
|
connectionScope.launch {
|
|
fragments.forEach { fragment ->
|
|
broadcastSinglePacket(RoutedPacket(fragment), gattServer, characteristic)
|
|
// 20ms delay between fragments (matching iOS/Rust)
|
|
delay(200)
|
|
}
|
|
}
|
|
return
|
|
}
|
|
}
|
|
|
|
// Send single packet if no fragmentation needed
|
|
broadcastSinglePacket(routed, gattServer, characteristic)
|
|
}
|
|
|
|
|
|
/**
|
|
* Public entry point for broadcasting - submits request to actor for serialization
|
|
*/
|
|
fun broadcastSinglePacket(
|
|
routed: RoutedPacket,
|
|
gattServer: BluetoothGattServer?,
|
|
characteristic: BluetoothGattCharacteristic?
|
|
) {
|
|
// Submit broadcast request to actor for serialized processing
|
|
broadcasterScope.launch {
|
|
try {
|
|
broadcasterActor.send(BroadcastRequest(routed, gattServer, characteristic))
|
|
} catch (e: Exception) {
|
|
Log.w(TAG, "Failed to send broadcast request to actor: ${e.message}")
|
|
// Fallback to direct processing if actor fails
|
|
broadcastSinglePacketInternal(routed, gattServer, characteristic)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Internal broadcast implementation - runs in serialized actor context
|
|
*/
|
|
private suspend fun broadcastSinglePacketInternal(
|
|
routed: RoutedPacket,
|
|
gattServer: BluetoothGattServer?,
|
|
characteristic: BluetoothGattCharacteristic?
|
|
) {
|
|
val packet = routed.packet
|
|
val data = packet.toBinaryData() ?: return
|
|
|
|
if (packet.recipientID != SpecialRecipients.BROADCAST) {
|
|
val recipientID = packet.recipientID?.let {
|
|
String(it).replace("\u0000", "").trim()
|
|
} ?: ""
|
|
|
|
// Try to find the recipient in server connections (subscribedDevices)
|
|
val targetDevice = connectionTracker.getSubscribedDevices()
|
|
.firstOrNull { connectionTracker.addressPeerMap[it.address] == recipientID }
|
|
|
|
// If found, send directly
|
|
if (targetDevice != null) {
|
|
Log.d(TAG, "Send packet type ${packet.type} directly to target device for recipient $recipientID: ${targetDevice.address}")
|
|
if (notifyDevice(targetDevice, data, gattServer, characteristic))
|
|
return // Sent, no need to continue
|
|
}
|
|
|
|
// Try to find the recipient in client connections (connectedDevices)
|
|
val targetDeviceConn = connectionTracker.getConnectedDevices().values
|
|
.firstOrNull { connectionTracker.addressPeerMap[it.device.address] == recipientID }
|
|
|
|
// If found, send directly
|
|
if (targetDeviceConn != null) {
|
|
Log.d(TAG, "Send packet type ${packet.type} directly to target client connection for recipient $recipientID: ${targetDeviceConn.device.address}")
|
|
if (writeToDeviceConn(targetDeviceConn, data))
|
|
return // Sent, no need to continue
|
|
}
|
|
}
|
|
|
|
// Else, continue with broadcasting to all devices
|
|
val subscribedDevices = connectionTracker.getSubscribedDevices()
|
|
val connectedDevices = connectionTracker.getConnectedDevices()
|
|
|
|
Log.i(TAG, "Broadcasting packet type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
|
|
|
|
val senderID = String(packet.senderID).replace("\u0000", "")
|
|
|
|
// Send to server connections (devices connected to our GATT server)
|
|
subscribedDevices.forEach { device ->
|
|
if (device.address == routed.relayAddress) {
|
|
Log.d(TAG, "Skipping broadcast back to relayer: ${device.address}")
|
|
return@forEach
|
|
}
|
|
if (connectionTracker.addressPeerMap[device.address] == senderID) {
|
|
Log.d(TAG, "Skipping broadcast back to sender: ${device.address}")
|
|
return@forEach
|
|
}
|
|
notifyDevice(device, data, gattServer, characteristic)
|
|
}
|
|
|
|
// Send to client connections
|
|
connectedDevices.values.forEach { deviceConn ->
|
|
if (deviceConn.isClient && deviceConn.gatt != null && deviceConn.characteristic != null) {
|
|
if (deviceConn.device.address == routed.relayAddress) {
|
|
Log.d(TAG, "Skipping broadcast back to relayer: ${deviceConn.device.address}")
|
|
return@forEach
|
|
}
|
|
if (connectionTracker.addressPeerMap[deviceConn.device.address] == senderID) {
|
|
Log.d(TAG, "Skipping broadcast back to sender: ${deviceConn.device.address}")
|
|
return@forEach
|
|
}
|
|
writeToDeviceConn(deviceConn, data)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Send data to a single device (server side)
|
|
*/
|
|
private fun notifyDevice(
|
|
device: BluetoothDevice,
|
|
data: ByteArray,
|
|
gattServer: BluetoothGattServer?,
|
|
characteristic: BluetoothGattCharacteristic?
|
|
): Boolean {
|
|
return try {
|
|
characteristic?.let { char ->
|
|
char.value = data
|
|
val result = gattServer?.notifyCharacteristicChanged(device, char, false) ?: false
|
|
result
|
|
} ?: false
|
|
} catch (e: Exception) {
|
|
Log.w(TAG, "Error sending to server connection ${device.address}: ${e.message}")
|
|
connectionScope.launch {
|
|
delay(CLEANUP_DELAY)
|
|
connectionTracker.removeSubscribedDevice(device)
|
|
connectionTracker.addressPeerMap.remove(device.address)
|
|
}
|
|
false
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Send data to a single device (client side)
|
|
*/
|
|
private fun writeToDeviceConn(
|
|
deviceConn: BluetoothConnectionTracker.DeviceConnection,
|
|
data: ByteArray
|
|
): Boolean {
|
|
return try {
|
|
deviceConn.characteristic?.let { char ->
|
|
char.value = data
|
|
val result = deviceConn.gatt?.writeCharacteristic(char) ?: false
|
|
result
|
|
} ?: false
|
|
} catch (e: Exception) {
|
|
Log.w(TAG, "Error sending to client connection ${deviceConn.device.address}: ${e.message}")
|
|
connectionScope.launch {
|
|
delay(CLEANUP_DELAY)
|
|
connectionTracker.cleanupDeviceConnection(deviceConn.device.address)
|
|
}
|
|
false
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get debug information
|
|
*/
|
|
fun getDebugInfo(): String {
|
|
return buildString {
|
|
appendLine("=== Packet Broadcaster Debug Info ===")
|
|
appendLine("Broadcaster Scope Active: ${broadcasterScope.isActive}")
|
|
appendLine("Actor Channel Closed: ${broadcasterActor.isClosedForSend}")
|
|
appendLine("Connection Scope Active: ${connectionScope.isActive}")
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Shutdown the broadcaster actor gracefully
|
|
*/
|
|
fun shutdown() {
|
|
Log.d(TAG, "Shutting down BluetoothPacketBroadcaster actor")
|
|
|
|
// Close the actor gracefully
|
|
broadcasterActor.close()
|
|
|
|
// Cancel the broadcaster scope
|
|
broadcasterScope.cancel()
|
|
|
|
Log.d(TAG, "BluetoothPacketBroadcaster shutdown complete")
|
|
}
|
|
} |