fix(ble): Add get_peer_mtu method to LinuxBluetoothDriver
## 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 <noreply@anthropic.com>
This commit is contained in:
parent
88bb2fc3fa
commit
2a34efc5c5
1 changed files with 19 additions and 0 deletions
|
|
@ -1088,6 +1088,25 @@ class LinuxBluetoothDriver(BLEDriverInterface):
|
||||||
return self._peers[address].connection_type
|
return self._peers[address].connection_type
|
||||||
return None
|
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):
|
def set_service_discovery_delay(self, seconds: float):
|
||||||
"""Set delay between connection and service discovery."""
|
"""Set delay between connection and service discovery."""
|
||||||
self.service_discovery_delay = seconds
|
self.service_discovery_delay = seconds
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue