From 3d50f1da239a35c5ce7496021b768a38c5c15ca8 Mon Sep 17 00:00:00 2001 From: CC Date: Mon, 8 Jun 2026 23:49:33 +0200 Subject: [PATCH] more robust --- .../android/mesh/BluetoothMeshService.kt | 2 +- .../mesh/BluetoothPacketBroadcaster.kt | 137 +++----------- .../android/mesh/MeshConnectionTracker.kt | 3 + .../java/com/bitchat/android/mesh/MeshCore.kt | 2 +- .../android/protocol/BinaryProtocol.kt | 4 + .../android/service/MeshForegroundService.kt | 6 + .../wifi-aware/WifiAwareConnectionTracker.kt | 41 ++++- .../android/wifi-aware/WifiAwareController.kt | 68 +++++-- .../wifi-aware/WifiAwareMeshService.kt | 168 +++++++++++++----- 9 files changed, 252 insertions(+), 179 deletions(-) diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt index eb14201d..618b66c6 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt @@ -869,7 +869,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic // Create NOISE_ENCRYPTED packet (not FILE_TRANSFER!) val packet = BitchatPacket( - version = 1u, + version = if (encrypted.size > 0xFFFF) 2u else 1u, type = MessageType.NOISE_ENCRYPTED.value, senderID = hexStringToByteArray(myPeerID), recipientID = hexStringToByteArray(recipientPeerID), diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt index 37af030d..efb3c4df 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt @@ -18,8 +18,6 @@ import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.Job -import java.util.concurrent.ConcurrentHashMap import kotlinx.coroutines.channels.actor /** @@ -117,7 +115,7 @@ class BluetoothPacketBroadcaster( // Actor scope for the broadcaster private val broadcasterScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) - private val transferJobs = ConcurrentHashMap() + private val fragmentingSender = FragmentingPacketSender(connectionScope, fragmentManager, TAG) // SERIALIZATION: Actor to serialize all broadcast operations @OptIn(kotlinx.coroutines.ObsoleteCoroutinesApi::class) @@ -139,71 +137,14 @@ class BluetoothPacketBroadcaster( gattServer: BluetoothGattServer?, characteristic: BluetoothGattCharacteristic? ) { - val packet = routed.packet - val isFile = packet.type == MessageType.FILE_TRANSFER.value - if (isFile) { - Log.d(TAG, "📤 Broadcasting FILE_TRANSFER: ${packet.payload.size} bytes") - } - // Prefer caller-provided transferId (e.g., for encrypted media), else derive for FILE_TRANSFER - val transferId = routed.transferId ?: (if (isFile) sha256Hex(packet.payload) else null) - // Check if we need to fragment - if (fragmentManager != null) { - val fragments = try { - fragmentManager.createFragments(packet) - } catch (e: Exception) { - Log.e(TAG, "❌ Fragment creation failed: ${e.message}", e) - if (isFile) { - Log.e(TAG, "❌ File fragmentation failed for ${packet.payload.size} byte file") - } - return - } - if (fragments.size > 1) { - if (isFile) { - Log.d(TAG, "🔀 File needs ${fragments.size} fragments") - } - Log.d(TAG, "Fragmenting packet into ${fragments.size} fragments") - if (transferId != null) { - TransferProgressManager.start(transferId, fragments.size) - } - val job = connectionScope.launch { - var sent = 0 - fragments.forEach { fragment -> - if (!isActive) return@launch - // If cancelled, stop sending remaining fragments - if (transferId != null && transferJobs[transferId]?.isCancelled == true) return@launch - broadcastSinglePacket(RoutedPacket(fragment, transferId = transferId), gattServer, characteristic) - // 20ms delay between fragments - delay(20) - if (transferId != null) { - sent += 1 - TransferProgressManager.progress(transferId, sent, fragments.size) - if (sent == fragments.size) TransferProgressManager.complete(transferId, fragments.size) - } - } - } - if (transferId != null) { - transferJobs[transferId] = job - job.invokeOnCompletion { transferJobs.remove(transferId) } - } - return - } - } - - // Send single packet if no fragmentation needed - if (transferId != null) { - TransferProgressManager.start(transferId, 1) - } - broadcastSinglePacket(routed, gattServer, characteristic) - if (transferId != null) { - TransferProgressManager.progress(transferId, 1, 1) - TransferProgressManager.complete(transferId, 1) + fragmentingSender.send(routed, "BLE broadcast") { packet -> + broadcastSinglePacket(packet, gattServer, characteristic) + true } } fun cancelTransfer(transferId: String): Boolean { - val job = transferJobs.remove(transferId) ?: return false - job.cancel() - return true + return fragmentingSender.cancelTransfer(transferId) } /** @@ -215,6 +156,18 @@ class BluetoothPacketBroadcaster( targetPeerID: String, gattServer: BluetoothGattServer?, characteristic: BluetoothGattCharacteristic? + ): Boolean { + if (!hasPeerConnection(targetPeerID)) return false + return fragmentingSender.send(routed, "BLE peer ${targetPeerID.take(8)}") { packet -> + sendSinglePacketToPeer(packet, targetPeerID, gattServer, characteristic) + } + } + + private fun sendSinglePacketToPeer( + routed: RoutedPacket, + targetPeerID: String, + gattServer: BluetoothGattServer?, + characteristic: BluetoothGattCharacteristic? ): Boolean { val packet = routed.packet val data = packet.toBinaryData() ?: return false @@ -222,11 +175,6 @@ class BluetoothPacketBroadcaster( if (isFile) { Log.d(TAG, "📤 Broadcasting FILE_TRANSFER: ${packet.payload.size} bytes") } - // Prefer caller-provided transferId (e.g., for encrypted media), else derive for FILE_TRANSFER - val transferId = routed.transferId ?: (if (isFile) sha256Hex(packet.payload) else null) - if (transferId != null) { - TransferProgressManager.start(transferId, 1) - } val typeName = MessageType.fromValue(packet.type)?.name ?: packet.type.toString() val senderPeerID = routed.peerID ?: packet.senderID.toHexString() val incomingAddr = routed.relayAddress @@ -241,10 +189,6 @@ class BluetoothPacketBroadcaster( if (serverTarget != null) { if (notifyDevice(serverTarget, data, gattServer, characteristic)) { logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, serverTarget.address, packet.ttl, packet.version, routeInfo) - if (transferId != null) { - TransferProgressManager.progress(transferId, 1, 1) - TransferProgressManager.complete(transferId, 1) - } return true } } @@ -255,10 +199,6 @@ class BluetoothPacketBroadcaster( if (clientTarget != null) { if (writeToDeviceConn(clientTarget, data)) { logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, clientTarget.device.address, packet.ttl, packet.version, routeInfo) - if (transferId != null) { - TransferProgressManager.progress(transferId, 1, 1) - TransferProgressManager.complete(transferId, 1) - } return true } } @@ -266,12 +206,6 @@ class BluetoothPacketBroadcaster( return false } - private fun sha256Hex(bytes: ByteArray): String = try { - val md = java.security.MessageDigest.getInstance("SHA-256") - md.update(bytes) - md.digest().joinToString("") { "%02x".format(it) } - } catch (_: Exception) { bytes.size.toString(16) } - /** * Public entry point for broadcasting - submits request to actor for serialization @@ -303,34 +237,19 @@ class BluetoothPacketBroadcaster( gattServer: BluetoothGattServer?, characteristic: BluetoothGattCharacteristic? ): Boolean { - val packet = routed.packet - val data = packet.toBinaryData() ?: return false - val typeName = MessageType.fromValue(packet.type)?.name ?: packet.type.toString() - val senderPeerID = routed.peerID ?: packet.senderID.toHexString() - val incomingAddr = routed.relayAddress - val incomingPeer = incomingAddr?.let { connectionTracker.addressPeerMap[it] } - val senderNick = senderPeerID.let { pid -> nicknameResolver?.invoke(pid) } - - // Try server-side connections first - val targetDevice = connectionTracker.getSubscribedDevices() - .firstOrNull { connectionTracker.addressPeerMap[it.address] == targetPeerID } - if (targetDevice != null) { - if (notifyDevice(targetDevice, data, gattServer, characteristic)) { - logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, targetDevice.address, packet.ttl) - return true - } + if (!hasPeerConnection(targetPeerID)) return false + return fragmentingSender.send(routed, "BLE peer ${targetPeerID.take(8)}") { packet -> + sendSinglePacketToPeer(packet, targetPeerID, gattServer, characteristic) } + } - // Try client-side connections next - val targetConn = connectionTracker.getConnectedDevices().values - .firstOrNull { connectionTracker.addressPeerMap[it.device.address] == targetPeerID } - if (targetConn != null) { - if (writeToDeviceConn(targetConn, data)) { - logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, targetConn.device.address, packet.ttl) - return true - } - } - return false + private fun hasPeerConnection(targetPeerID: String): Boolean { + val hasServerTarget = connectionTracker.getSubscribedDevices() + .any { connectionTracker.addressPeerMap[it.address] == targetPeerID } + if (hasServerTarget) return true + + return connectionTracker.getConnectedDevices().values + .any { connectionTracker.addressPeerMap[it.device.address] == targetPeerID } } /** diff --git a/app/src/main/java/com/bitchat/android/mesh/MeshConnectionTracker.kt b/app/src/main/java/com/bitchat/android/mesh/MeshConnectionTracker.kt index 01df23c8..0dd6a552 100644 --- a/app/src/main/java/com/bitchat/android/mesh/MeshConnectionTracker.kt +++ b/app/src/main/java/com/bitchat/android/mesh/MeshConnectionTracker.kt @@ -1,6 +1,7 @@ package com.bitchat.android.mesh import android.util.Log +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.delay import kotlinx.coroutines.launch @@ -131,6 +132,8 @@ abstract class MeshConnectionTracker( if (expired.isNotEmpty()) { Log.d(tag, "Cleaned up ${expired.size} expired connection attempts") } + } catch (e: CancellationException) { + break } catch (e: Exception) { Log.w(tag, "Error in periodic cleanup: ${e.message}") } diff --git a/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt b/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt index fbd64876..0abea334 100644 --- a/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt +++ b/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt @@ -465,7 +465,7 @@ class MeshCore( val np = NoisePayload(type = NoisePayloadType.FILE_TRANSFER, data = tlv).encode() val enc = encryptionService.encrypt(np, recipientPeerID) val packet = BitchatPacket( - version = 1u, + version = if (enc.size > 0xFFFF) 2u else 1u, type = MessageType.NOISE_ENCRYPTED.value, senderID = MeshPacketUtils.hexStringToByteArray(myPeerID), recipientID = MeshPacketUtils.hexStringToByteArray(recipientPeerID), diff --git a/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt b/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt index db17ba1c..88e5f783 100644 --- a/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt +++ b/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt @@ -255,6 +255,10 @@ object BinaryProtocol { if (packet.version >= 2u.toUByte()) { buffer.putInt(payloadDataSize) // 4 bytes for v2+ } else { + if (payloadDataSize > 0xFFFF || (originalPayloadSize ?: 0) > 0xFFFF) { + Log.w("BinaryProtocol", "Cannot encode oversized v1 packet payload: $payloadDataSize bytes") + return null + } buffer.putShort(payloadDataSize.toShort()) // 2 bytes for v1 } diff --git a/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt b/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt index 2d1a4d9c..c1ba78c9 100644 --- a/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt +++ b/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt @@ -226,6 +226,12 @@ class MeshForegroundService : Service() { private fun ensureMeshStarted() { if (isShuttingDown) return + try { + com.bitchat.android.wifiaware.WifiAwareController.startIfPossible() + } catch (e: Exception) { + android.util.Log.e("MeshForegroundService", "Failed to ensure Wi-Fi Aware transport: ${e.message}") + } + val bleEnabled = try { com.bitchat.android.ui.debug.DebugPreferenceManager.getBleEnabled(true) } catch (_: Exception) { diff --git a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareConnectionTracker.kt b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareConnectionTracker.kt index c07feb34..ddfe0a53 100644 --- a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareConnectionTracker.kt +++ b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareConnectionTracker.kt @@ -48,17 +48,20 @@ class WifiAwareConnectionTracker( // Ensure any pending/active network request is explicitly released releaseNetworkRequest(canonicalId) + removePendingConnection(id) + removePendingConnection(canonicalId) } fun releaseNetworkRequest(id: String) { - if (!networkCallbacks.containsKey(id)) return + val canonicalId = resolveCanonicalPeerId(id) + if (!networkCallbacks.containsKey(canonicalId)) return // 3. Unregister network callback properly from ConnectivityManager - networkCallbacks.remove(id)?.let { + networkCallbacks.remove(canonicalId)?.let { try { - Log.d(TAG, "Unregistering network callback for $id") + Log.d(TAG, "Unregistering network callback for $canonicalId") cm.unregisterNetworkCallback(it) - } catch (e: Exception) { Log.w(TAG, "Error unregistering callback for $id: ${e.message}") } + } catch (e: Exception) { Log.w(TAG, "Error unregistering callback for $canonicalId: ${e.message}") } } } @@ -125,11 +128,37 @@ class WifiAwareConnectionTracker( } fun addServerSocket(peerId: String, socket: ServerSocket) { - serverSockets[peerId] = socket + val canonicalId = resolveCanonicalPeerId(peerId) + serverSockets.put(canonicalId, socket)?.let { + try { it.close() } catch (e: Exception) { Log.w(TAG, "Error closing replaced server socket for $peerId: ${e.message}") } + } + } + + fun hasOpenServerSocket(peerId: String): Boolean { + val canonicalId = resolveCanonicalPeerId(peerId) + val socket = serverSockets[canonicalId] ?: return false + if (!socket.isClosed) return true + serverSockets.remove(canonicalId) + return false + } + + fun closeServerSocket(peerId: String) { + val canonicalId = resolveCanonicalPeerId(peerId) + serverSockets.remove(canonicalId)?.let { + try { it.close() } catch (e: Exception) { Log.w(TAG, "Error closing server socket for $peerId: ${e.message}") } + } } fun addNetworkCallback(peerId: String, callback: ConnectivityManager.NetworkCallback) { - networkCallbacks[peerId] = callback + val canonicalId = resolveCanonicalPeerId(peerId) + networkCallbacks.put(canonicalId, callback)?.let { + try { + Log.d(TAG, "Replacing network callback for $canonicalId") + cm.unregisterNetworkCallback(it) + } catch (e: Exception) { + Log.w(TAG, "Error unregistering replaced callback for $canonicalId: ${e.message}") + } + } } /** diff --git a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareController.kt b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareController.kt index 58641f66..447cf903 100644 --- a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareController.kt +++ b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareController.kt @@ -23,6 +23,8 @@ object WifiAwareController { private var service: WifiAwareMeshService? = null private var appContext: Context? = null + private val lifecycleLock = Any() + private var starting = false private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) @@ -71,23 +73,33 @@ object WifiAwareController { } fun startIfPossible() { - if (_running.value) { + val reusableService = synchronized(lifecycleLock) { if (!_enabled.value) return - // If already marked running but we are calling this (e.g. after a drop), - // ensure we clean up the old service before restarting. - stop() + val existing = service + if (existing?.isRunning() == true) { + _running.value = true + return + } + if (starting) return + starting = true + existing + } + + val ctx = appContext ?: run { + synchronized(lifecycleLock) { starting = false } + return } - if (!_enabled.value) return - val ctx = appContext ?: return if (Build.VERSION.SDK_INT < Build.VERSION_CODES.Q) { Log.w(TAG, "Wi‑Fi Aware requires Android 10 (Q)+; disabled.") try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Wi‑Fi Aware not supported on this device (requires Android 10+)")) } catch (_: Exception) {} + synchronized(lifecycleLock) { starting = false } return } val awareManager = ctx.getSystemService(android.net.wifi.aware.WifiAwareManager::class.java) if (awareManager == null || !awareManager.isAvailable) { Log.w(TAG, "Wi-Fi Aware is not currently available; not starting") try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Wi‑Fi Aware is not available on this device right now")) } catch (_: Exception) {} + synchronized(lifecycleLock) { starting = false } return } @@ -104,6 +116,7 @@ object WifiAwareController { if (!locationEnabled) { Log.w(TAG, "Location services are disabled; Wi-Fi Aware cannot start.") try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Enable Location Services to start Wi-Fi Aware")) } catch (_: Exception) {} + synchronized(lifecycleLock) { starting = false } return } @@ -113,34 +126,54 @@ object WifiAwareController { if (!granted) { Log.w(TAG, "Missing NEARBY_WIFI_DEVICES permission; not starting Wi‑Fi Aware") try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Grant Nearby Wi‑Fi Devices to start Wi‑Fi Aware")) } catch (_: Exception) {} + synchronized(lifecycleLock) { starting = false } return } } + if (!_enabled.value) { + synchronized(lifecycleLock) { starting = false } + return + } try { - Log.i(TAG, "Instantiating WifiAwareMeshService...") - val startedService = WifiAwareMeshService(ctx) + val startedService = reusableService ?: run { + Log.i(TAG, "Instantiating WifiAwareMeshService...") + WifiAwareMeshService(ctx) + } startedService.startServices() if (startedService.isRunning()) { - service = startedService - _running.value = true + synchronized(lifecycleLock) { + service = startedService + _running.value = true + } try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Wi‑Fi Aware started")) } catch (_: Exception) {} } else { - try { startedService.stopServices() } catch (_: Exception) { } - service = null - _running.value = false + if (reusableService == null) { + try { startedService.stopServices() } catch (_: Exception) { } + } + synchronized(lifecycleLock) { + if (service === startedService) service = null + _running.value = false + } try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Wi‑Fi Aware did not start")) } catch (_: Exception) {} } } catch (e: Throwable) { Log.e(TAG, "Failed to start WifiAwareMeshService", e) _running.value = false try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Wi‑Fi Aware failed to start: ${e.message}")) } catch (_: Exception) {} + } finally { + synchronized(lifecycleLock) { starting = false } } } fun stop() { - try { service?.stopServices() } catch (_: Exception) { } - service = null - _running.value = false + val stopped = synchronized(lifecycleLock) { + val current = service + service = null + starting = false + _running.value = false + current + } + try { stopped?.stopServices() } catch (_: Exception) { } try { com.bitchat.android.services.AppStateStore.clearTransportPeers("WIFI") } catch (_: Exception) { } _connectedPeers.value = emptyMap() _knownPeers.value = emptyMap() @@ -149,7 +182,8 @@ object WifiAwareController { } internal fun onServiceStopped(stoppedService: WifiAwareMeshService) { - if (service === stoppedService) { + synchronized(lifecycleLock) { + if (service !== stoppedService) return service = null _running.value = false try { com.bitchat.android.services.AppStateStore.clearTransportPeers("WIFI") } catch (_: Exception) { } diff --git a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt index 3bb4e470..4eab93b3 100644 --- a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt +++ b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt @@ -12,6 +12,7 @@ import android.util.Log import androidx.annotation.RequiresApi import androidx.annotation.RequiresPermission import com.bitchat.android.crypto.EncryptionService +import com.bitchat.android.mesh.FragmentingPacketSender import com.bitchat.android.mesh.MeshCore import com.bitchat.android.mesh.MeshService import com.bitchat.android.mesh.MeshTransport @@ -25,6 +26,7 @@ import com.bitchat.android.protocol.SpecialRecipients import com.bitchat.android.service.TransportBridgeService import com.bitchat.android.sync.GossipSyncManager import com.bitchat.android.util.toHexString +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob @@ -40,6 +42,7 @@ import java.nio.ByteBuffer import java.nio.ByteOrder import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicInteger /** * WifiAware mesh service - LATEST @@ -69,6 +72,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor private val serviceScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) private val wifiTransport = WifiAwareTransport() private lateinit var meshCore: MeshCore + private lateinit var fragmentingSender: FragmentingPacketSender // Service-level notification manager for background (no-UI) DMs private val serviceNotificationManager = com.bitchat.android.ui.NotificationManager( @@ -84,6 +88,8 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor private var subscribeSession: SubscribeDiscoverySession? = null private val listenerExec = Executors.newCachedThreadPool() private var isActive = false + @Volatile private var recoveryInProgress = false + private val sessionGeneration = AtomicInteger(0) // Delegate override var delegate: WifiAwareMeshDelegate? = null @@ -139,6 +145,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor } ) ) + fragmentingSender = FragmentingPacketSender(serviceScope, meshCore.fragmentManager, TAG) } private fun handleMessageReceived(message: BitchatMessage) { @@ -206,7 +213,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor val packet = routed.packet if (packet.senderID.toHexString() == myPeerID && !packet.route.isNullOrEmpty()) { val firstHop = packet.route!![0].toHexString() - if (sendPacketToPeer(firstHop, packet)) { + if (sendRoutedPacketToPeer(firstHop, routed)) { Log.d(TAG, "TX: source-routed packet sent only to first Wi-Fi hop ${firstHop.take(8)}") return } @@ -215,15 +222,15 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor val recipientId = packet.recipientID?.toHexString() if (recipientId != null && !packet.recipientID.contentEquals(SpecialRecipients.BROADCAST)) { - if (sendPacketToPeer(recipientId, packet)) { + if (sendRoutedPacketToPeer(recipientId, routed)) { Log.d(TAG, "TX: addressed packet sent directly to Wi-Fi peer ${recipientId.take(8)}") return } } - // Wi-Fi Aware uses full packets; no fragmentation - val data = packet.toBinaryData() ?: return - serviceScope.launch { broadcastRaw(data) } + fragmentingSender.send(routed, "Wi-Fi Aware broadcast") { single -> + broadcastSinglePacket(single) + } } // Expose a public method so BLE can forward relays to Wi-Fi Aware @@ -235,22 +242,40 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor * Send packet to connected peer. */ private fun sendPacketToPeer(peerID: String, packet: BitchatPacket): Boolean { - // Wi-Fi Aware uses full packets; no fragmentation + return sendRoutedPacketToPeer(peerID, RoutedPacket(packet)) + } + + private fun sendRoutedPacketToPeer(peerID: String, routed: RoutedPacket): Boolean { + if (connectionTracker.getSocketForPeer(peerID) == null) { + Log.w(TAG, "TX: no socket for ${peerID.take(8)}") + return false + } + return fragmentingSender.send(routed, "Wi-Fi Aware peer ${peerID.take(8)}") { single -> + sendSinglePacketToPeer(peerID, single.packet) + } + } + + private fun broadcastSinglePacket(routed: RoutedPacket): Boolean { + val data = routed.packet.toBinaryData() ?: return false + broadcastRaw(data) + return true + } + + private fun sendSinglePacketToPeer(peerID: String, packet: BitchatPacket): Boolean { val data = packet.toBinaryData() ?: return false val sock = connectionTracker.getSocketForPeer(peerID) if (sock == null) { Log.w(TAG, "TX: no socket for ${peerID.take(8)}") return false } - serviceScope.launch { - try { - sock.write(data) - Log.d(TAG, "TX: packet type=${packet.type} to ${peerID.take(8)} (bytes=${data.size})") - } catch (e: IOException) { - Log.e(TAG, "TX: write to ${peerID.take(8)} failed: ${e.message}") - } + try { + sock.write(data) + Log.d(TAG, "TX: packet type=${packet.type} to ${peerID.take(8)} (bytes=${data.size})") + return true + } catch (e: IOException) { + Log.e(TAG, "TX: write to ${peerID.take(8)} failed: ${e.message}") + return false } - return true } @@ -273,12 +298,17 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor Log.i(TAG, "Wi-Fi Aware transport disabled by debug settings; not starting") return } + if (recoveryInProgress) { + Log.i(TAG, "Wi-Fi Aware recovery cleanup still in progress; deferring start") + return + } val manager = awareManager if (manager == null || !manager.isAvailable) { Log.w(TAG, "Wi-Fi Aware manager unavailable; not starting") return } isActive = true + val generation = sessionGeneration.incrementAndGet() Log.i(TAG, "Starting Wi-Fi Aware mesh with peer ID: $myPeerID") manager.attach(object : AttachCallback() { @@ -288,6 +318,10 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor Manifest.permission.NEARBY_WIFI_DEVICES ]) override fun onAttached(session: WifiAwareSession) { + if (!isCurrentSession(generation)) { + session.close() + return + } wifiAwareSession = session Log.i(TAG, "Wi-Fi Aware attached; starting publish & subscribe (peerID=$myPeerID)") @@ -299,6 +333,10 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor .build(), object : DiscoverySessionCallback() { override fun onPublishStarted(pub: PublishDiscoverySession) { + if (!isCurrentSession(generation)) { + pub.close() + return + } publishSession = pub Log.d(TAG, "PUBLISH: onPublishStarted()") try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Wi-Fi Aware Publish Started")) } catch (_: Exception) {} @@ -308,6 +346,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor serviceSpecificInfo: ByteArray, matchFilter: List ) { + if (!isCurrentSession(generation)) return val peerId = try { String(serviceSpecificInfo) } catch (_: Exception) { "" } handleToPeerId[peerHandle] = peerId if (peerId.isNotBlank()) { @@ -322,6 +361,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor peerHandle: PeerHandle, message: ByteArray ) { + if (!isCurrentSession(generation)) return if (message.isEmpty()) return val subscriberId = try { String(message) } catch (_: Exception) { "" } if (subscriberId == myPeerID) return @@ -333,10 +373,11 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor } override fun onSessionTerminated() { + if (!isCurrentSession(generation)) return Log.e(TAG, "PUBLISH: onSessionTerminated()") publishSession = null val shouldRestart = isActive && com.bitchat.android.wifiaware.WifiAwareController.enabled.value - handleUnexpectedStop() + handleUnexpectedStop(generation) if (shouldRestart) { Log.i(TAG, "PUBLISH: Scheduling Wi-Fi Aware restart") com.bitchat.android.wifiaware.WifiAwareController.restartIfStillEnabled(2000) @@ -353,6 +394,10 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor .build(), object : DiscoverySessionCallback() { override fun onSubscribeStarted(sub: SubscribeDiscoverySession) { + if (!isCurrentSession(generation)) { + sub.close() + return + } subscribeSession = sub Log.d(TAG, "SUBSCRIBE: onSubscribeStarted()") try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().addDebugMessage(com.bitchat.android.ui.debug.DebugMessage.SystemMessage("Wi-Fi Aware Subscribe Started")) } catch (_: Exception) {} @@ -362,6 +407,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor serviceSpecificInfo: ByteArray, matchFilter: List ) { + if (!isCurrentSession(generation)) return val peerId = try { String(serviceSpecificInfo) } catch (_: Exception) { "" } handleToPeerId[peerHandle] = peerId val msgId = (System.nanoTime() and 0x7fffffff).toInt() @@ -375,6 +421,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor peerHandle: PeerHandle, message: ByteArray ) { + if (!isCurrentSession(generation)) return if (message.isEmpty()) return val peerId = handleToPeerId[peerHandle] ?: return if (peerId == myPeerID) return @@ -384,10 +431,11 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor } override fun onSessionTerminated() { + if (!isCurrentSession(generation)) return Log.e(TAG, "SUBSCRIBE: onSessionTerminated()") subscribeSession = null val shouldRestart = isActive && com.bitchat.android.wifiaware.WifiAwareController.enabled.value - handleUnexpectedStop() + handleUnexpectedStop(generation) if (shouldRestart) { Log.i(TAG, "SUBSCRIBE: Scheduling Wi-Fi Aware restart") com.bitchat.android.wifiaware.WifiAwareController.restartIfStillEnabled(2000) @@ -398,15 +446,20 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor ) } override fun onAttachFailed() { + if (!isCurrentSession(generation)) return Log.e(TAG, "Wi-Fi Aware attach failed") - handleUnexpectedStop() + handleUnexpectedStop(generation) + if (com.bitchat.android.wifiaware.WifiAwareController.enabled.value) { + com.bitchat.android.wifiaware.WifiAwareController.restartIfStillEnabled(3000) + } } override fun onAwareSessionTerminated() { + if (!isCurrentSession(generation)) return Log.e(TAG, "Aware Session Terminated unexpectedly") wifiAwareSession = null val shouldRestart = com.bitchat.android.wifiaware.WifiAwareController.enabled.value - handleUnexpectedStop() + handleUnexpectedStop(generation) if (shouldRestart) { com.bitchat.android.wifiaware.WifiAwareController.restartIfStillEnabled(3000) } @@ -427,6 +480,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor override fun stopServices() { val wasActive = isActive isActive = false + sessionGeneration.incrementAndGet() Log.i(TAG, "Stopping Wi-Fi Aware mesh") // Unregister from bridge @@ -456,27 +510,38 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor } } - private fun handleUnexpectedStop() { + private fun isCurrentSession(generation: Int): Boolean { + return generation == sessionGeneration.get() && isActive + } + + private fun handleUnexpectedStop(generation: Int) { + if (generation != sessionGeneration.get()) return if (!isActive) { - com.bitchat.android.wifiaware.WifiAwareController.onServiceStopped(this) return } + recoveryInProgress = true isActive = false TransportBridgeService.unregister("WIFI") try { com.bitchat.android.services.AppStateStore.clearTransportPeers("WIFI") } catch (_: Exception) { } + val oldPublishSession = publishSession + val oldSubscribeSession = subscribeSession + val oldWifiAwareSession = wifiAwareSession serviceScope.launch { - try { meshCore.stopCore() } catch (_: Exception) { } - try { connectionTracker.stop() } catch (_: Exception) { } - try { publishSession?.close() } catch (_: Exception) { } - try { subscribeSession?.close() } catch (_: Exception) { } - try { wifiAwareSession?.close() } catch (_: Exception) { } - publishSession = null - subscribeSession = null - wifiAwareSession = null - handleToPeerId.clear() - try { meshCore.shutdown() } catch (_: Exception) { } - com.bitchat.android.wifiaware.WifiAwareController.onServiceStopped(this@WifiAwareMeshService) - serviceScope.cancel() + try { + try { meshCore.stopCore() } catch (_: Exception) { } + try { connectionTracker.stop() } catch (_: Exception) { } + try { oldPublishSession?.close() } catch (_: Exception) { } + try { oldSubscribeSession?.close() } catch (_: Exception) { } + try { oldWifiAwareSession?.close() } catch (_: Exception) { } + if (generation == sessionGeneration.get() && !isActive) { + if (publishSession === oldPublishSession) publishSession = null + if (subscribeSession === oldSubscribeSession) subscribeSession = null + if (wifiAwareSession === oldWifiAwareSession) wifiAwareSession = null + handleToPeerId.clear() + } + } finally { + recoveryInProgress = false + } } } @@ -511,16 +576,16 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor if (!connectionTracker.isConnectionAttemptAllowed(peerId)) continue Log.i(TAG, "🔄 Maintenance: attempting reconnect to ${peerId.take(8)}") - if (connectionTracker.addPendingConnection(peerId)) { - // Resend ping to trigger handshake - val msgId = (System.nanoTime() and 0x7fffffff).toInt() - try { - subscribeSession?.sendMessage(handle, msgId, myPeerID.toByteArray()) - } catch (e: Exception) { - Log.w(TAG, "Failed to send maintenance ping to ${peerId.take(8)}: ${e.message}") - } + // Resend ping to trigger a fresh server-ready/data-path attempt. + val msgId = (System.nanoTime() and 0x7fffffff).toInt() + try { + subscribeSession?.sendMessage(handle, msgId, myPeerID.toByteArray()) + } catch (e: Exception) { + Log.w(TAG, "Failed to send maintenance ping to ${peerId.take(8)}: ${e.message}") } } + } catch (e: CancellationException) { + break } catch (e: Exception) { Log.e(TAG, "Error in connection maintenance: ${e.message}") } @@ -542,16 +607,23 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor val peerId = handleToPeerId[peerHandle] ?: return if (!amIServerFor(peerId)) return - if (connectionTracker.serverSockets.containsKey(peerId)) { + if (connectionTracker.hasOpenServerSocket(peerId)) { Log.v(TAG, "↪ already serving $peerId, skipping") return } + if (!connectionTracker.addPendingConnection(peerId)) { + return + } val ss = ServerSocket() try { ss.reuseAddress = true ss.bind(java.net.InetSocketAddress(0)) - } catch (e: Exception) { Log.e(TAG, "Failed to bind server socket", e) } + } catch (e: Exception) { + Log.e(TAG, "Failed to bind server socket", e) + handleNetworkFailure(peerId) + return + } connectionTracker.addServerSocket(peerId, ss) val port = ss.localPort @@ -597,6 +669,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor serviceScope.launch { delay(150); sendBroadcastAnnounce() } } catch (ioe: IOException) { Log.e(TAG, "SERVER: accept failed for ${peerId.take(8)}", ioe) + handleNetworkFailure(peerId) } } @@ -687,6 +760,9 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor Log.v(TAG, "↪ already client-connected to $peerId, skipping") return } + if (!connectionTracker.addPendingConnection(peerId)) { + return + } val port = ByteBuffer.wrap(payload).order(ByteOrder.BIG_ENDIAN).int Log.i(TAG, "CLIENT: Received server-ready from ${peerId.take(8)} on port $port. Requesting network...") @@ -758,6 +834,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor serviceScope.launch { delay(150); sendBroadcastAnnounce() } } catch (ioe: IOException) { Log.e(TAG, "CLIENT: socket connect failed to ${peerId.take(8)}", ioe) + handleNetworkFailure(peerId) } } override fun onLost(network: Network) { @@ -867,11 +944,9 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor private fun handleNetworkFailure(peerId: String) { serviceScope.launch { Log.d(TAG, "Network failure cleanup for: $peerId") - // Specifically release the callback if it didn't happen automatically - connectionTracker.releaseNetworkRequest(peerId) - if (!connectionTracker.isConnected(peerId)) { val canonicalPeerId = connectionTracker.canonicalPeerId(peerId) + connectionTracker.disconnect(peerId) meshCore.removePeer(canonicalPeerId) if (canonicalPeerId != peerId) { meshCore.removePeer(peerId) @@ -1133,6 +1208,9 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor override fun sendPacketToPeer(peerID: String, packet: BitchatPacket): Boolean { return this@WifiAwareMeshService.sendPacketToPeer(peerID, packet) } + override fun cancelTransfer(transferId: String): Boolean { + return fragmentingSender.cancelTransfer(transferId) + } override fun getDeviceAddressForPeer(peerID: String): String? { return connectionTracker.getSocketForPeer(peerID)?.let { resolveScopedAddress(it.rawSocket) } }