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 #766 from zxzxwu/l2cap

Browse Source 
Remove depreacated L2CAP APIs
This commit is contained in:
zxzxwu
2025-08-28 10:58:35 +08:00
committed by GitHub
parent 8bda7d2212 6091e6365d
commit 6f73b736d7
4 changed files with 19 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
bumble/avrcp.py
View File

@@ -1147,7 +1147,9 @@ class Protocol(utils.EventEmitter):
A 'connection' event will be emitted when a connection is made, and a 'start' A 'connection' event will be emitted when a connection is made, and a 'start'
event will be emitted when the protocol is ready to be used on that connection. event will be emitted when the protocol is ready to be used on that connection.
""" """
device.register_l2cap_server(avctp.AVCTP_PSM, self._on_avctp_connection) device.create_l2cap_server(
l2cap.ClassicChannelSpec(avctp.AVCTP_PSM), self._on_avctp_connection
)
async def connect(self, connection: Connection) -> None: async def connect(self, connection: Connection) -> None:
""" """

40
bumble/device.py
View File

@@ -1869,16 +1869,6 @@ class Connection(utils.CompositeEventEmitter):
def send_l2cap_pdu(self, cid: int, pdu: bytes) -> None: def send_l2cap_pdu(self, cid: int, pdu: bytes) -> None:
self.device.send_l2cap_pdu(self.handle, cid, pdu) self.device.send_l2cap_pdu(self.handle, cid, pdu)
@utils.deprecated("Please use create_l2cap_channel()")
async def open_l2cap_channel(
self,
psm,
max_credits=DEVICE_DEFAULT_L2CAP_COC_MAX_CREDITS,
mtu=DEVICE_DEFAULT_L2CAP_COC_MTU,
mps=DEVICE_DEFAULT_L2CAP_COC_MPS,
):
return await self.device.open_l2cap_channel(self, psm, max_credits, mtu, mps)
@overload @overload
async def create_l2cap_channel( async def create_l2cap_channel(
self, spec: l2cap.ClassicChannelSpec self, spec: l2cap.ClassicChannelSpec
@@ -2591,36 +2581,6 @@ class Device(utils.CompositeEventEmitter):
None, None,
) )
@utils.deprecated("Please use create_l2cap_server()")
def register_l2cap_server(self, psm, server) -> int:
return self.l2cap_channel_manager.register_server(psm, server)
@utils.deprecated("Please use create_l2cap_server()")
def register_l2cap_channel_server(
self,
psm,
server,
max_credits=DEVICE_DEFAULT_L2CAP_COC_MAX_CREDITS,
mtu=DEVICE_DEFAULT_L2CAP_COC_MTU,
mps=DEVICE_DEFAULT_L2CAP_COC_MPS,
):
return self.l2cap_channel_manager.register_le_coc_server(
psm, server, max_credits, mtu, mps
)
@utils.deprecated("Please use create_l2cap_channel()")
async def open_l2cap_channel(
self,
connection,
psm,
max_credits=DEVICE_DEFAULT_L2CAP_COC_MAX_CREDITS,
mtu=DEVICE_DEFAULT_L2CAP_COC_MTU,
mps=DEVICE_DEFAULT_L2CAP_COC_MPS,
):
return await self.l2cap_channel_manager.open_le_coc(
connection, psm, max_credits, mtu, mps
)
@overload @overload
async def create_l2cap_channel( async def create_l2cap_channel(
self, self,

24
bumble/hid.py
View File

@@ -217,33 +217,41 @@ class HID(ABC, utils.EventEmitter):
self.role = role self.role = role
# Register ourselves with the L2CAP channel manager # Register ourselves with the L2CAP channel manager
device.register_l2cap_server(HID_CONTROL_PSM, self.on_l2cap_connection) device.create_l2cap_server(
device.register_l2cap_server(HID_INTERRUPT_PSM, self.on_l2cap_connection) l2cap.ClassicChannelSpec(HID_CONTROL_PSM), self.on_l2cap_connection
)
device.create_l2cap_server(
l2cap.ClassicChannelSpec(HID_INTERRUPT_PSM), self.on_l2cap_connection
)
device.on(device.EVENT_CONNECTION, self.on_device_connection) device.on(device.EVENT_CONNECTION, self.on_device_connection)
async def connect_control_channel(self) -> None: async def connect_control_channel(self) -> None:
if not self.connection:
raise InvalidStateError("Connection is not established!")
# Create a new L2CAP connection - control channel # Create a new L2CAP connection - control channel
try: try:
channel = await self.device.l2cap_channel_manager.connect( channel = await self.connection.create_l2cap_channel(
self.connection, HID_CONTROL_PSM l2cap.ClassicChannelSpec(HID_CONTROL_PSM)
) )
channel.sink = self.on_ctrl_pdu channel.sink = self.on_ctrl_pdu
self.l2cap_ctrl_channel = channel self.l2cap_ctrl_channel = channel
except ProtocolError: except ProtocolError:
logging.exception(f'L2CAP connection failed.') logging.exception('L2CAP connection failed.')
raise raise
async def connect_interrupt_channel(self) -> None: async def connect_interrupt_channel(self) -> None:
if not self.connection:
raise InvalidStateError("Connection is not established!")
# Create a new L2CAP connection - interrupt channel # Create a new L2CAP connection - interrupt channel
try: try:
channel = await self.device.l2cap_channel_manager.connect( channel = await self.connection.create_l2cap_channel(
self.connection, HID_INTERRUPT_PSM l2cap.ClassicChannelSpec(HID_CONTROL_PSM)
) )
channel.sink = self.on_intr_pdu channel.sink = self.on_intr_pdu
self.l2cap_intr_channel = channel self.l2cap_intr_channel = channel
except ProtocolError: except ProtocolError:
logging.exception(f'L2CAP connection failed.') logging.exception('L2CAP connection failed.')
raise raise
async def disconnect_interrupt_channel(self) -> None: async def disconnect_interrupt_channel(self) -> None:

60
bumble/l2cap.py
View File

@@ -1531,16 +1531,6 @@ class ChannelManager:
if cid in self.fixed_channels: if cid in self.fixed_channels:
del self.fixed_channels[cid] del self.fixed_channels[cid]
@utils.deprecated("Please use create_classic_server")
def register_server(
self,
psm: int,
server: Callable[[ClassicChannel], Any],
) -> int:
return self.create_classic_server(
handler=server, spec=ClassicChannelSpec(psm=psm)
).psm
def create_classic_server( def create_classic_server(
self, self,
spec: ClassicChannelSpec, spec: ClassicChannelSpec,
@@ -1577,22 +1567,6 @@ class ChannelManager:
return self.servers[spec.psm] return self.servers[spec.psm]
@utils.deprecated("Please use create_le_credit_based_server()")
def register_le_coc_server(
self,
psm: int,
server: Callable[[LeCreditBasedChannel], Any],
max_credits: int,
mtu: int,
mps: int,
) -> int:
return self.create_le_credit_based_server(
spec=LeCreditBasedChannelSpec(
psm=None if psm == 0 else psm, mtu=mtu, mps=mps, max_credits=max_credits
),
handler=server,
).psm
def create_le_credit_based_server( def create_le_credit_based_server(
self, self,
spec: LeCreditBasedChannelSpec, spec: LeCreditBasedChannelSpec,
@@ -2145,17 +2119,6 @@ class ChannelManager:
if channel.source_cid in connection_channels: if channel.source_cid in connection_channels:
del connection_channels[channel.source_cid] del connection_channels[channel.source_cid]
@utils.deprecated("Please use create_le_credit_based_channel()")
async def open_le_coc(
self, connection: Connection, psm: int, max_credits: int, mtu: int, mps: int
) -> LeCreditBasedChannel:
return await self.create_le_credit_based_channel(
connection=connection,
spec=LeCreditBasedChannelSpec(
psm=psm, max_credits=max_credits, mtu=mtu, mps=mps
),
)
async def create_le_credit_based_channel( async def create_le_credit_based_channel(
self, self,
connection: Connection, connection: Connection,
@@ -2202,12 +2165,6 @@ class ChannelManager:
return channel return channel
@utils.deprecated("Please use create_classic_channel()")
async def connect(self, connection: Connection, psm: int) -> ClassicChannel:
return await self.create_classic_channel(
connection=connection, spec=ClassicChannelSpec(psm=psm)
)
async def create_classic_channel( async def create_classic_channel(
self, connection: Connection, spec: ClassicChannelSpec self, connection: Connection, spec: ClassicChannelSpec
) -> ClassicChannel: ) -> ClassicChannel:
@@ -2244,20 +2201,3 @@ class ChannelManager:
raise e raise e
return channel return channel
# -----------------------------------------------------------------------------
# Deprecated Classes
# -----------------------------------------------------------------------------
class Channel(ClassicChannel):
@utils.deprecated("Please use ClassicChannel")
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
class LeConnectionOrientedChannel(LeCreditBasedChannel):
@utils.deprecated("Please use LeCreditBasedChannel")
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)