mirror of
https://github.com/permissionlesstech/bitchat-android.git
synced 2026-07-25 06:25:21 +00:00
Implement Noise XX Handshake Protocol for Direct Messages (#180)
* noise * works? * noise * temporary * better * wip: use subnet * better * barely working * werk * subnet * fix peer ID * 8 byte peer ID * wip noise * wip fixes for noise * std lib for noise * noise handshake one step further * buffers * use fork * fix imports * simplify counter * remove trash * hashing * no prologue * nice * wip, use noise encryption * peer ID hex * simplify session manager * heavy logging * use singleton * Fix Noise session race condition with elegant per-peer actor serialization - Use Kotlin coroutine actors for per-peer packet processing - Each peer gets dedicated actor that processes packets sequentially - Eliminates race conditions in session management without complex locking - Single surgical change in PacketProcessor - minimal, maintainable - Leverages Kotlin's native concurrency primitives * decrypt correctly * iniator works now * clean code and fix signature to null * better * no signature in private message * small fixes * refactor ack * refactor but untested * messages working * wip ack * wip fix ack * more logging * pending tracker * keep pending connections on errors * less logging * refactor model * refactor frombinarydata * idendityannouncement refactor and update to new binary protocol * fix keys * refix keys * dms work * revert to mainnet * do not change bluetooth adapter name * keep code but uncomment * clean up comments * cleanup comments
This commit is contained in:
callebtc
@@ -8,11 +8,20 @@ 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
|
||||
* 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,
|
||||
@@ -25,14 +34,40 @@ class 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
|
||||
// Check if we need to fragment
|
||||
if (fragmentManager != null) {
|
||||
val fragments = fragmentManager.createFragments(packet)
|
||||
if (fragments.size > 1) {
|
||||
@@ -41,7 +76,7 @@ class BluetoothPacketBroadcaster(
|
||||
fragments.forEach { fragment ->
|
||||
broadcastSinglePacket(RoutedPacket(fragment), gattServer, characteristic)
|
||||
// 20ms delay between fragments (matching iOS/Rust)
|
||||
delay(20)
|
||||
delay(200)
|
||||
}
|
||||
}
|
||||
return
|
||||
@@ -54,12 +89,32 @@ class BluetoothPacketBroadcaster(
|
||||
|
||||
|
||||
/**
|
||||
* Broadcast single packet to connected devices with connection limit enforcement
|
||||
* 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
|
||||
@@ -96,7 +151,7 @@ class BluetoothPacketBroadcaster(
|
||||
val subscribedDevices = connectionTracker.getSubscribedDevices()
|
||||
val connectedDevices = connectionTracker.getConnectedDevices()
|
||||
|
||||
Log.d(TAG, "Broadcasting packet type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
|
||||
Log.i(TAG, "Broadcasting packet type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
|
||||
|
||||
val senderID = String(packet.senderID).replace("\u0000", "")
|
||||
|
||||
@@ -177,4 +232,31 @@ class BluetoothPacketBroadcaster(
|
||||
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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user