mirror of
https://github.com/permissionlesstech/bitchat-android.git
synced 2026-07-25 00:25:20 +00:00
Merge branch 'gossip-routing-2' of https://github.com/callebtc/bitchat-android into gossip-routing-2
This commit is contained in:
Vendored
+5
@@ -26,3 +26,8 @@
|
||||
-keepnames class org.torproject.arti.**
|
||||
-dontwarn info.guardianproject.arti.**
|
||||
-dontwarn org.torproject.arti.**
|
||||
|
||||
# Fix for AbstractMethodError on API < 29 where LocationListener methods are abstract
|
||||
-keepclassmembers class * implements android.location.LocationListener {
|
||||
public <methods>;
|
||||
}
|
||||
|
||||
@@ -109,11 +109,19 @@ class BluetoothConnectionManager(
|
||||
}
|
||||
// Connection caps: enforce on change
|
||||
connectionScope.launch {
|
||||
dbg.maxConnectionsOverall.collect {
|
||||
dbg.maxConnectionsOverall.collect { maxOverall ->
|
||||
if (!isActive) return@collect
|
||||
// 1. Enforce client limits (handled by tracker)
|
||||
connectionTracker.enforceConnectionLimits()
|
||||
// Also enforce server side best-effort
|
||||
serverManager.enforceServerLimit(dbg.maxServerConnections.value)
|
||||
|
||||
// 2. Enforce overall limit on server connections if needed
|
||||
// (Tracker knows about all connections but can't disconnect servers directly)
|
||||
val maxServer = dbg.maxServerConnections.value
|
||||
val excessServers = connectionTracker.getExcessServerConnections(maxServer, maxOverall)
|
||||
excessServers.forEach { device ->
|
||||
Log.d(TAG, "Disconnecting server ${device.address} due to overall cap")
|
||||
serverManager.disconnectDevice(device)
|
||||
}
|
||||
}
|
||||
}
|
||||
connectionScope.launch {
|
||||
@@ -123,9 +131,18 @@ class BluetoothConnectionManager(
|
||||
}
|
||||
}
|
||||
connectionScope.launch {
|
||||
dbg.maxServerConnections.collect {
|
||||
dbg.maxServerConnections.collect { maxServer ->
|
||||
if (!isActive) return@collect
|
||||
serverManager.enforceServerLimit(dbg.maxServerConnections.value)
|
||||
// Enforce server specific limit
|
||||
serverManager.enforceServerLimit(maxServer)
|
||||
|
||||
// Also check if this change puts us over the overall limit
|
||||
val maxOverall = dbg.maxConnectionsOverall.value
|
||||
val excessServers = connectionTracker.getExcessServerConnections(maxServer, maxOverall)
|
||||
excessServers.forEach { device ->
|
||||
Log.d(TAG, "Disconnecting server ${device.address} due to overall cap")
|
||||
serverManager.disconnectDevice(device)
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (_: Exception) { }
|
||||
|
||||
@@ -229,11 +229,17 @@ class BluetoothConnectionTracker(
|
||||
*/
|
||||
fun getConnectedDeviceCount(): Int = connectedDevices.size
|
||||
|
||||
/**
|
||||
* Check if connection limit is reached
|
||||
*/
|
||||
/**
|
||||
* Check if connection limit is reached
|
||||
*/
|
||||
fun isConnectionLimitReached(): Boolean {
|
||||
return connectedDevices.size >= powerManager.getMaxConnections()
|
||||
// Respect debug override if set
|
||||
val dbg = try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance() } catch (_: Exception) { null }
|
||||
val maxConnections = dbg?.maxConnectionsOverall?.value ?: powerManager.getMaxConnections()
|
||||
return connectedDevices.size >= maxConnections
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -244,10 +250,9 @@ class BluetoothConnectionTracker(
|
||||
val dbg = try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance() } catch (_: Exception) { null }
|
||||
val maxOverall = dbg?.maxConnectionsOverall?.value ?: powerManager.getMaxConnections()
|
||||
val maxClient = dbg?.maxClientConnections?.value ?: maxOverall
|
||||
val maxServer = dbg?.maxServerConnections?.value ?: maxOverall
|
||||
// Note: maxServer is handled by GattServerManager, but we need to respect overall limit here too
|
||||
|
||||
val clients = connectedDevices.values.filter { it.isClient }
|
||||
val servers = connectedDevices.values.filter { !it.isClient }
|
||||
|
||||
// Enforce client cap first (we can actively disconnect)
|
||||
if (clients.size > maxClient) {
|
||||
@@ -259,22 +264,56 @@ class BluetoothConnectionTracker(
|
||||
}
|
||||
}
|
||||
|
||||
// Note: server cap enforced in GattServerManager (we don't have server handle here)
|
||||
|
||||
// Enforce overall cap by disconnecting oldest client connections
|
||||
// Re-check overall cap after client cleanup
|
||||
if (connectedDevices.size > maxOverall) {
|
||||
Log.i(TAG, "Enforcing overall cap: ${connectedDevices.size} > $maxOverall")
|
||||
val excess = connectedDevices.size - maxOverall
|
||||
val toDisconnect = connectedDevices.values
|
||||
.filter { it.isClient } // only clients from here
|
||||
|
||||
// Prefer disconnecting clients first to satisfy overall cap
|
||||
val clientsToDisconnect = connectedDevices.values
|
||||
.filter { it.isClient }
|
||||
.sortedBy { it.connectedAt }
|
||||
.take(excess)
|
||||
toDisconnect.forEach { dc ->
|
||||
|
||||
clientsToDisconnect.forEach { dc ->
|
||||
Log.d(TAG, "Disconnecting client ${dc.device.address} due to overall cap")
|
||||
dc.gatt?.disconnect()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get excess server connections that should be disconnected to satisfy limits.
|
||||
* This allows the Manager to coordinate server disconnects since Tracker doesn't control the server.
|
||||
*/
|
||||
fun getExcessServerConnections(maxServer: Int, maxOverall: Int): List<BluetoothDevice> {
|
||||
val servers = connectedDevices.values.filter { !it.isClient }
|
||||
val excessList = mutableListOf<BluetoothDevice>()
|
||||
|
||||
// 1. Check server specific limit
|
||||
if (servers.size > maxServer) {
|
||||
val excessCount = servers.size - maxServer
|
||||
val toRemove = servers.sortedBy { it.connectedAt }.take(excessCount)
|
||||
excessList.addAll(toRemove.map { it.device })
|
||||
}
|
||||
|
||||
// 2. Check overall limit (considering we might have already removed some above)
|
||||
// We need to count how many connections we will have after the above removals
|
||||
val currentTotal = connectedDevices.size
|
||||
val plannedRemovals = excessList.size
|
||||
val projectedTotal = currentTotal - plannedRemovals
|
||||
|
||||
if (projectedTotal > maxOverall) {
|
||||
val furtherExcess = projectedTotal - maxOverall
|
||||
// We can only remove servers here. Clients are handled in enforceConnectionLimits.
|
||||
// Filter out devices we already planned to remove
|
||||
val remainingServers = servers.filter { s -> excessList.none { it.address == s.device.address } }
|
||||
val toRemove = remainingServers.sortedBy { it.connectedAt }.take(furtherExcess)
|
||||
excessList.addAll(toRemove.map { it.device })
|
||||
}
|
||||
|
||||
return excessList
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up a specific device connection
|
||||
|
||||
@@ -49,15 +49,28 @@ class BluetoothGattServerManager(
|
||||
fun enforceServerLimit(maxServer: Int) {
|
||||
if (maxServer <= 0) return
|
||||
try {
|
||||
val subs = connectionTracker.getSubscribedDevices()
|
||||
if (subs.size > maxServer) {
|
||||
val excess = subs.size - maxServer
|
||||
subs.take(excess).forEach { d ->
|
||||
try { gattServer?.cancelConnection(d) } catch (_: Exception) { }
|
||||
// Use connection tracker to get actual connected server devices
|
||||
val servers = connectionTracker.getConnectedDevices().values.filter { !it.isClient }
|
||||
if (servers.size > maxServer) {
|
||||
val excess = servers.size - maxServer
|
||||
// Disconnect oldest
|
||||
servers.sortedBy { it.connectedAt }.take(excess).forEach { d ->
|
||||
try { gattServer?.cancelConnection(d.device) } catch (_: Exception) { }
|
||||
}
|
||||
}
|
||||
} catch (_: Exception) { }
|
||||
}
|
||||
|
||||
/**
|
||||
* Disconnect a specific device (used by ConnectionManager to enforce overall limits)
|
||||
*/
|
||||
fun disconnectDevice(device: BluetoothDevice) {
|
||||
try {
|
||||
gattServer?.cancelConnection(device)
|
||||
} catch (e: Exception) {
|
||||
Log.w(TAG, "Error disconnecting device ${device.address}: ${e.message}")
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start GATT server
|
||||
@@ -122,9 +135,10 @@ class BluetoothGattServerManager(
|
||||
|
||||
// Try to cancel any active connections explicitly before closing
|
||||
try {
|
||||
val devices = connectionTracker.getSubscribedDevices()
|
||||
devices.forEach { d ->
|
||||
try { gattServer?.cancelConnection(d) } catch (_: Exception) { }
|
||||
// Disconnect ALL server connections
|
||||
val servers = connectionTracker.getConnectedDevices().values.filter { !it.isClient }
|
||||
servers.forEach { d ->
|
||||
try { gattServer?.cancelConnection(d.device) } catch (_: Exception) { }
|
||||
}
|
||||
} catch (_: Exception) { }
|
||||
|
||||
|
||||
@@ -378,7 +378,7 @@ class BluetoothPacketBroadcaster(
|
||||
val subscribedDevices = connectionTracker.getSubscribedDevices()
|
||||
val connectedDevices = connectionTracker.getConnectedDevices()
|
||||
|
||||
Log.i(TAG, "Broadcasting packet type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
|
||||
Log.i(TAG, "Broadcasting packet v${packet.version} type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
|
||||
|
||||
val senderID = String(packet.senderID).replace("\u0000", "")
|
||||
|
||||
|
||||
@@ -77,8 +77,31 @@ class FragmentManager {
|
||||
val fragmentID = FragmentPayload.generateFragmentID()
|
||||
|
||||
// iOS: stride(from: 0, to: fullData.count, by: maxFragmentSize)
|
||||
val fragmentChunks = stride(0, fullData.size, MAX_FRAGMENT_SIZE) { offset ->
|
||||
val endOffset = minOf(offset + MAX_FRAGMENT_SIZE, fullData.size)
|
||||
// Calculate dynamic fragment size to fit in MTU (512)
|
||||
// Packet = Header + Sender + Recipient + Route + FragmentHeader + Payload + PaddingBuffer
|
||||
val hasRoute = packet.route != null
|
||||
val version = if (hasRoute) 2 else 1
|
||||
val headerSize = if (version == 2) 15 else 13
|
||||
val senderSize = 8
|
||||
val recipientSize = if (packet.recipientID != null) 8 else 0
|
||||
// Route: 1 byte count + 8 bytes per hop
|
||||
val routeSize = if (hasRoute) (1 + (packet.route?.size ?: 0) * 8) else 0
|
||||
val fragmentHeaderSize = 13 // FragmentPayload header
|
||||
val paddingBuffer = 16 // MessagePadding.optimalBlockSize adds 16 bytes overhead
|
||||
|
||||
// 512 - Overhead
|
||||
val packetOverhead = headerSize + senderSize + recipientSize + routeSize + fragmentHeaderSize + paddingBuffer
|
||||
val maxDataSize = (512 - packetOverhead).coerceAtMost(MAX_FRAGMENT_SIZE)
|
||||
|
||||
if (maxDataSize <= 0) {
|
||||
Log.e(TAG, "❌ Calculated maxDataSize is non-positive ($maxDataSize). Route too large?")
|
||||
return emptyList()
|
||||
}
|
||||
|
||||
Log.d(TAG, "📏 Dynamic fragment size: $maxDataSize (MAX: $MAX_FRAGMENT_SIZE, Overhead: $packetOverhead)")
|
||||
|
||||
val fragmentChunks = stride(0, fullData.size, maxDataSize) { offset ->
|
||||
val endOffset = minOf(offset + maxDataSize, fullData.size)
|
||||
fullData.sliceArray(offset..<endOffset)
|
||||
}
|
||||
|
||||
|
||||
@@ -219,7 +219,10 @@ object BinaryProtocol {
|
||||
val signatureBytes = if (packet.signature != null) SIGNATURE_SIZE else 0
|
||||
val sizeFieldBytes = if (isCompressed) (if (packet.version >= 2u.toUByte()) 4 else 2) else 0
|
||||
val payloadBytes = payload.size + sizeFieldBytes
|
||||
val capacity = headerSize + SENDER_ID_SIZE + recipientBytes + payloadBytes + signatureBytes + 16 // small slack
|
||||
val routeBytes = if (!packet.route.isNullOrEmpty() && packet.version >= 2u.toUByte()) {
|
||||
1 + (packet.route!!.size.coerceAtMost(255) * SENDER_ID_SIZE)
|
||||
} else 0
|
||||
val capacity = headerSize + SENDER_ID_SIZE + recipientBytes + payloadBytes + signatureBytes + routeBytes + 16 // small slack
|
||||
val buffer = ByteBuffer.allocate(capacity.coerceAtLeast(512)).apply { order(ByteOrder.BIG_ENDIAN) }
|
||||
|
||||
// Header
|
||||
@@ -368,9 +371,16 @@ object BinaryProtocol {
|
||||
var routeCount = 0
|
||||
if (hasRoute) {
|
||||
// Peek count (1 byte) without consuming buffer for now
|
||||
val mark = buffer.position()
|
||||
if (raw.size >= mark + 1) {
|
||||
routeCount = raw[mark].toUByte().toInt()
|
||||
// The buffer is currently positioned at the start of SenderID (after fixed header)
|
||||
// We must skip SenderID and RecipientID (if present) to find the route count
|
||||
val currentPos = buffer.position()
|
||||
var routeOffset = currentPos + SENDER_ID_SIZE
|
||||
if (hasRecipient) {
|
||||
routeOffset += RECIPIENT_ID_SIZE
|
||||
}
|
||||
|
||||
if (raw.size >= routeOffset + 1) {
|
||||
routeCount = raw[routeOffset].toUByte().toInt()
|
||||
}
|
||||
expectedSize += 1 + (routeCount * SENDER_ID_SIZE)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
package com.bitchat.android.mesh
|
||||
|
||||
import com.bitchat.android.protocol.BitchatPacket
|
||||
import com.bitchat.android.protocol.MessageType
|
||||
import com.bitchat.android.model.FragmentPayload
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNotNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import org.junit.runner.RunWith
|
||||
import org.robolectric.RobolectricTestRunner
|
||||
import java.util.Random
|
||||
|
||||
@RunWith(RobolectricTestRunner::class)
|
||||
class FragmentManagerTest {
|
||||
|
||||
private lateinit var fragmentManager: FragmentManager
|
||||
private val senderID = "1122334455667788"
|
||||
private val recipientID = "8877665544332211"
|
||||
|
||||
@Before
|
||||
fun setup() {
|
||||
fragmentManager = FragmentManager()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `test fragmentation without route`() {
|
||||
// Create a large payload (e.g., 1000 bytes)
|
||||
val payload = ByteArray(1000)
|
||||
Random().nextBytes(payload)
|
||||
|
||||
val packet = BitchatPacket(
|
||||
version = 1u,
|
||||
type = MessageType.MESSAGE.value,
|
||||
senderID = hexStringToByteArray(senderID),
|
||||
recipientID = hexStringToByteArray(recipientID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = 7u,
|
||||
route = null
|
||||
)
|
||||
|
||||
val fragments = fragmentManager.createFragments(packet)
|
||||
|
||||
assertTrue("Should create multiple fragments", fragments.size > 1)
|
||||
|
||||
// Verify each fragment fits in MTU (512)
|
||||
for (fragment in fragments) {
|
||||
val encodedSize = fragment.toBinaryData()?.size ?: 0
|
||||
assertTrue("Fragment encoded size should be <= 512, was $encodedSize", encodedSize <= 512)
|
||||
|
||||
// Inspect the payload data size
|
||||
val fragmentPayload = FragmentPayload.decode(fragment.payload)
|
||||
assertNotNull(fragmentPayload)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `test fragmentation with route`() {
|
||||
// Create a large payload
|
||||
val payload = ByteArray(1000)
|
||||
Random().nextBytes(payload)
|
||||
|
||||
// Create a fake route (3 hops)
|
||||
val route = listOf(
|
||||
hexStringToByteArray("AABBCCDDEEFF0011"),
|
||||
hexStringToByteArray("1100FFEEDDCCBBAA"),
|
||||
hexStringToByteArray("1234567890ABCDEF")
|
||||
)
|
||||
|
||||
val packet = BitchatPacket(
|
||||
version = 2u,
|
||||
type = MessageType.MESSAGE.value,
|
||||
senderID = hexStringToByteArray(senderID),
|
||||
recipientID = hexStringToByteArray(recipientID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = 7u,
|
||||
route = route
|
||||
)
|
||||
|
||||
val fragments = fragmentManager.createFragments(packet)
|
||||
|
||||
assertTrue("Should create multiple fragments", fragments.size > 1)
|
||||
|
||||
// Verify fragments retain the route and version 2
|
||||
for (fragment in fragments) {
|
||||
assertEquals("Fragment version should be 2", 2u.toUByte(), fragment.version)
|
||||
assertEquals("Fragment should have the route", route.size, fragment.route?.size)
|
||||
|
||||
val encodedSize = fragment.toBinaryData()?.size ?: 0
|
||||
assertTrue("Fragment encoded size should be <= 512, was $encodedSize", encodedSize <= 512)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `test fragmentation size difference with and without route`() {
|
||||
// This test specifically checks if the dynamic calculation logic works
|
||||
// by observing that fragments with routes carry less data payload per fragment
|
||||
|
||||
val payload = ByteArray(2000) // Large enough to ensure full fragments
|
||||
Random().nextBytes(payload)
|
||||
|
||||
// 1. Without route
|
||||
val packetNoRoute = BitchatPacket(
|
||||
version = 1u,
|
||||
type = MessageType.MESSAGE.value,
|
||||
senderID = hexStringToByteArray(senderID),
|
||||
recipientID = hexStringToByteArray(recipientID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = 7u,
|
||||
route = null
|
||||
)
|
||||
val fragmentsNoRoute = fragmentManager.createFragments(packetNoRoute)
|
||||
val firstFragPayloadNoRoute = FragmentPayload.decode(fragmentsNoRoute[0].payload)
|
||||
val dataSizeNoRoute = firstFragPayloadNoRoute?.data?.size ?: 0
|
||||
|
||||
// 2. With large route (e.g., 5 hops)
|
||||
val route = List(5) { hexStringToByteArray("000000000000000$it") }
|
||||
val packetWithRoute = BitchatPacket(
|
||||
version = 2u,
|
||||
type = MessageType.MESSAGE.value,
|
||||
senderID = hexStringToByteArray(senderID),
|
||||
recipientID = hexStringToByteArray(recipientID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = 7u,
|
||||
route = route
|
||||
)
|
||||
val fragmentsWithRoute = fragmentManager.createFragments(packetWithRoute)
|
||||
val firstFragPayloadWithRoute = FragmentPayload.decode(fragmentsWithRoute[0].payload)
|
||||
val dataSizeWithRoute = firstFragPayloadWithRoute?.data?.size ?: 0
|
||||
|
||||
println("Data size without route: $dataSizeNoRoute")
|
||||
println("Data size with route: $dataSizeWithRoute")
|
||||
|
||||
assertTrue("Data payload should be smaller with route", dataSizeWithRoute < dataSizeNoRoute)
|
||||
|
||||
// Rough verification of the math:
|
||||
// 5 hops * 8 bytes = 40 bytes extra.
|
||||
// Plus v2 header overhead differences.
|
||||
// The difference should be roughly 40+ bytes.
|
||||
assertTrue("Difference should be significant", (dataSizeNoRoute - dataSizeWithRoute) >= 40)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `test reassembly`() {
|
||||
val originalPayload = ByteArray(1500)
|
||||
Random().nextBytes(originalPayload)
|
||||
|
||||
val originalPacket = BitchatPacket(
|
||||
version = 1u,
|
||||
type = MessageType.FILE_TRANSFER.value,
|
||||
senderID = hexStringToByteArray(senderID),
|
||||
recipientID = hexStringToByteArray(recipientID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = originalPayload,
|
||||
ttl = 7u
|
||||
)
|
||||
|
||||
val fragments = fragmentManager.createFragments(originalPacket)
|
||||
|
||||
var reassembledPacket: BitchatPacket? = null
|
||||
|
||||
// Feed fragments back into FragmentManager
|
||||
// Note: FragmentManager stores state in incomingFragments
|
||||
|
||||
for (fragment in fragments) {
|
||||
val result = fragmentManager.handleFragment(fragment)
|
||||
if (result != null) {
|
||||
reassembledPacket = result
|
||||
}
|
||||
}
|
||||
|
||||
assertNotNull("Should have reassembled packet", reassembledPacket)
|
||||
assertEquals("Type should match", originalPacket.type, reassembledPacket!!.type)
|
||||
assertEquals("Payload size should match", originalPacket.payload.size, reassembledPacket.payload.size)
|
||||
assertTrue("Payload content should match", originalPacket.payload.contentEquals(reassembledPacket.payload))
|
||||
}
|
||||
|
||||
private fun hexStringToByteArray(hexString: String): ByteArray {
|
||||
val result = ByteArray(8)
|
||||
for (i in 0 until 8) {
|
||||
val byteStr = hexString.substring(i * 2, i * 2 + 2)
|
||||
result[i] = byteStr.toInt(16).toByte()
|
||||
}
|
||||
return result
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user