From 2a34efc5c56f9ecfbfdc95d4fcaa6f78e844c6d3 Mon Sep 17 00:00:00 2001 From: torlando-tech Date: Tue, 4 Nov 2025 01:04:03 -0500 Subject: [PATCH] fix(ble): Add get_peer_mtu method to LinuxBluetoothDriver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem The identity handshake handler called `self.driver.get_peer_mtu(address)`, but this method didn't exist, causing: ``` [Error] BLEInterface[BLE Interface] failed to process identity handshake from dev:B8:27:EB:A8:A7:22: 'LinuxBluetoothDriver' object has no attribute 'get_peer_mtu' ``` ## Solution Added `get_peer_mtu(address)` method to LinuxBluetoothDriver that: 1. Checks central connections (self._peers) for MTU when we're the central 2. Checks peripheral connections (gatt_server.connected_centrals) for MTU when we're the peripheral 3. Returns None if peer not found in either This mirrors the existing `get_peer_role()` pattern and provides thread-safe access to MTU information for both connection types. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/RNS/Interfaces/linux_bluetooth_driver.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/src/RNS/Interfaces/linux_bluetooth_driver.py b/src/RNS/Interfaces/linux_bluetooth_driver.py index c9275b3..3e74989 100644 --- a/src/RNS/Interfaces/linux_bluetooth_driver.py +++ b/src/RNS/Interfaces/linux_bluetooth_driver.py @@ -1088,6 +1088,25 @@ class LinuxBluetoothDriver(BLEDriverInterface): return self._peers[address].connection_type return None + def get_peer_mtu(self, address: str) -> Optional[int]: + """Return the negotiated MTU for a peer connection. + + Checks both central connections (we connected to them) and peripheral + connections (they connected to us). + """ + # Check central connections (we are central) + with self._peers_lock: + if address in self._peers: + return self._peers[address].mtu + + # Check peripheral connections (we are peripheral, they are central) + if self.gatt_server: + with self.gatt_server.centrals_lock: + if address in self.gatt_server.connected_centrals: + return self.gatt_server.connected_centrals[address].get("mtu") + + return None + def set_service_discovery_delay(self, seconds: float): """Set delay between connection and service discovery.""" self.service_discovery_delay = seconds