pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

Cleanup legacy L2CAP API

Browse Source
This commit is contained in:
Josh Wu
2023-10-11 13:49:02 +08:00
parent 7255a09705
commit 5a85765360
14 changed files with 98 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
bumble/l2cap.py
View File

@@ -1036,9 +1036,11 @@ class LeCreditBasedChannel(EventEmitter):
out_queue: Deque[bytes]
connection_result: Optional[asyncio.Future[LeCreditBasedChannel]]
disconnection_result: Optional[asyncio.Future[None]]
in_sdu: Optional[bytes]
out_sdu: Optional[bytes]
state: State
connection: Connection
sink: Optional[Callable[[bytes], Any]]
def __init__(
self,
@@ -1525,7 +1527,7 @@ class ChannelManager:
if cid in self.fixed_channels:
del self.fixed_channels[cid]
@deprecated("Please use create_classic_channel_server")
@deprecated("Please use create_classic_server")
def register_server(
self,
psm: int,
@@ -1540,7 +1542,7 @@ class ChannelManager:
spec: ClassicChannelSpec,
handler: Optional[Callable[[ClassicChannel], Any]] = None,
) -> ClassicChannelServer:
if spec.psm is None:
if not spec.psm:
# Find a free PSM
for candidate in range(
L2CAP_PSM_DYNAMIC_RANGE_START, L2CAP_PSM_DYNAMIC_RANGE_END + 1, 2
@@ -1592,7 +1594,7 @@ class ChannelManager:
spec: LeCreditBasedChannelSpec,
handler: Optional[Callable[[LeCreditBasedChannel], Any]] = None,
) -> LeCreditBasedChannelServer:
if spec.psm is None:
if not spec.psm:
# Find a free PSM
for candidate in range(
L2CAP_LE_PSM_DYNAMIC_RANGE_START, L2CAP_LE_PSM_DYNAMIC_RANGE_END + 1

7
bumble/profiles/asha_service.py
View File

@@ -19,6 +19,8 @@
import struct
import logging
from typing import List
from bumble import l2cap
from ..core import AdvertisingData
from ..device import Device, Connection
from ..gatt import (
@@ -149,7 +151,10 @@ class AshaService(TemplateService):
channel.sink = on_data
# let the server find a free PSM
self.psm = self.device.register_l2cap_channel_server(self.psm, on_coc, 8)
self.psm = device.create_l2cap_server(
spec=l2cap.LeCreditBasedChannelSpec(psm=self.psm, max_credits=8),
handler=on_coc,
).psm
self.le_psm_out_characteristic = Characteristic(
GATT_ASHA_LE_PSM_OUT_CHARACTERISTIC,
Characteristic.Properties.READ,

8
bumble/rfcomm.py
View File

@@ -898,8 +898,8 @@ class Client:
async def start(self) -> Multiplexer:
# Create a new L2CAP connection
try:
self.l2cap_channel = await self.device.l2cap_channel_manager.connect(
self.connection, RFCOMM_PSM
self.l2cap_channel = await self.connection.create_l2cap_channel(
spec=l2cap.ClassicChannelSpec(RFCOMM_PSM)
)
except ProtocolError as error:
logger.warning(f'L2CAP connection failed: {error}')
@@ -936,7 +936,9 @@ class Server(EventEmitter):
self.acceptors = {}
# Register ourselves with the L2CAP channel manager
device.register_l2cap_server(RFCOMM_PSM, self.on_connection)
device.create_l2cap_server(
spec=l2cap.ClassicChannelSpec(psm=RFCOMM_PSM), handler=self.on_connection
)
def listen(self, acceptor: Callable[[DLC], None], channel: int = 0) -> int:
if channel:

9
bumble/sdp.py
View File

@@ -766,8 +766,9 @@ class Client:
self.channel = None
async def connect(self, connection: Connection) -> None:
result = await self.device.l2cap_channel_manager.connect(connection, SDP_PSM)
self.channel = result
self.channel = await connection.create_l2cap_channel(
spec=l2cap.ClassicChannelSpec(SDP_PSM)
)
async def disconnect(self) -> None:
if self.channel:
@@ -933,7 +934,9 @@ class Server:
self.current_response = None
def register(self, l2cap_channel_manager: l2cap.ChannelManager) -> None:
l2cap_channel_manager.register_server(SDP_PSM, self.on_connection)
l2cap_channel_manager.create_classic_server(
spec=l2cap.ClassicChannelSpec(psm=SDP_PSM), handler=self.on_connection
)
def send_response(self, response):
logger.debug(f'{color(">>> Sending SDP Response", "blue")}: {response}')