auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-04-16 00:25:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #884 from timrid/fix-multiple-le-connections

Browse Source 
Connecting multiple times to a LE device is working correctly again
This commit is contained in:
Josh Wu
2026-02-06 11:25:44 +08:00
committed by GitHub
parent 377fa896f7 81d9adb983
commit d810d93aaf
2 changed files with 48 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
bumble/controller.py
View File

@@ -481,6 +481,18 @@ class Controller:
return connection
return None
def find_le_connection_by_handle(self, handle: int) -> Connection | None:
for connection in self.le_connections.values():
if connection.handle == handle:
return connection
return None
def find_classic_connection_by_handle(self, handle: int) -> Connection | None:
for connection in self.classic_connections.values():
if connection.handle == handle:
return connection
return None
def find_classic_sco_link_by_handle(self, handle: int) -> ScoLink | None:
for connection in self.sco_links.values():
if connection.handle == handle:
@@ -604,6 +616,8 @@ class Controller:
)
)
del self.le_connections[connection.peer_address]
def create_le_connection(self, peer_address: hci.Address) -> None:
'''
Called when we receive advertisement matching connection filter.
@@ -1201,22 +1215,23 @@ class Controller:
# Notify the link of the disconnection
handle = command.connection_handle
if connection := self.find_connection_by_handle(handle):
if connection := self.find_classic_connection_by_handle(handle):
if self.link:
if connection.transport == PhysicalTransport.BR_EDR:
self.send_lmp_packet(
connection.peer_address,
lmp.LmpDetach(command.reason),
)
self.on_classic_disconnected(
connection.peer_address, command.reason
)
else:
connection.send_ll_control_pdu(ll.TerminateInd(command.reason))
self.on_le_disconnected(connection, command.reason)
self.send_lmp_packet(
connection.peer_address,
lmp.LmpDetach(command.reason),
)
self.on_classic_disconnected(connection.peer_address, command.reason)
else:
# Remove the connection
del self.classic_connections[connection.peer_address]
elif connection := self.find_le_connection_by_handle(handle):
if self.link:
connection.send_ll_control_pdu(ll.TerminateInd(command.reason))
self.on_le_disconnected(connection, command.reason)
else:
# Remove the connection
del self.le_connections[connection.peer_address]
elif sco_link := self.find_classic_sco_link_by_handle(handle):
if self.link:
if (

21
tests/device_test.py
View File

@@ -309,6 +309,27 @@ async def test_legacy_advertising_disconnection(auto_restart):
assert not devices[0].is_advertising
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_le_multiple_connects():
devices = TwoDevices()
for controller in devices.controllers:
controller.le_features |= hci.LeFeatureMask.LE_EXTENDED_ADVERTISING
for dev in devices:
await dev.power_on()
await devices[0].start_advertising(auto_restart=True, advertising_interval_min=1.0)
connection = await devices[1].connect(devices[0].random_address)
await connection.disconnect()
await async_barrier()
await async_barrier()
# a second connection attempt is working
connection = await devices[1].connect(devices[0].random_address)
await connection.disconnect()
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_advertising_and_scanning():