Skip to content

Allow injecting a custom ChannelManager into Device #942

Description

@corby

Summary

ChannelManager.on_l2cap_connection_parameter_update_request (bumble/l2cap.py:2524-2558) unconditionally auto-accepts every L2CAP Connection Parameter Update Request the peripheral sends, and immediately follows up with HCI_LE_Connection_Update_Command using the peer's requested parameters. There is no way for application code to veto the request or to substitute its own parameters.

Motivation

I act as a GATT central talking to peripherals that, mid-session, send CPURs asking for a slower connection interval / higher latency. That kills throughput during bulk transfers I control from the central side. Today my only options are:

  • Monkey-patch ChannelManager on the live Device.
  • Fork bumble.
    Both are ugly. I'd like a first-class way to keep the connection parameters I negotiated at connect time.

Why I can't work around it cleanly today

  • Device.__init__ hard-codes the manager (bumble/device.py:2449):
    self.l2cap_channel_manager = l2cap.ChannelManager(
        config.l2cap_extended_features
    )
    so I can't pass in a subclass.
  • The existing Connection.EVENT_CONNECTION_PARAMETERS_UPDATE event fires after the update has already been issued, so it's too late to refuse.

Proposal

Make the ChannelManager class injectable on Device:

device = Device.with_hci(
    "my-central",
    Address(dongle_address),
    source, sink,
    channel_manager_cls=MyChannelManager,   # default: l2cap.ChannelManager
)

with the matching channel_manager_cls kwarg threaded through Device.__init__ and used at device.py:2449 instead of the hard-coded l2cap.ChannelManager. Default behavior is unchanged.
This is the right fix because it lets advanced users override any L2CAP signaling handler (not just CPUR) without bumble needing to grow a new flag or event every time someone hits a hard-coded policy decision.

Example usage

class MyChannelManager(l2cap.ChannelManager):
    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.ignore_conn_param_update_req = False

    def on_l2cap_connection_parameter_update_request(
        self,
        connection: Connection,
        cid: int,
        request: L2CAP_Connection_Parameter_Update_Request,
    ) -> None:
        if not self.ignore_conn_param_update_req:
             # Just bounce the request back as default accept
            return super().on_l2cap_connection_parameter_update_request(
                connection, cid, request
            )
        self.send_control_frame(
            connection,
            cid,
            L2CAP_Connection_Parameter_Update_Response(
                identifier=request.identifier,
                result=L2CAP_CONNECTION_PARAMETERS_REJECTED_RESULT,
            ),
        )

device = Device.with_hci(
    "my-central", Address(dongle_address), source, sink,
    channel_manager_cls=MyChannelManager,
)

Environment

  • bumble 0.0.230
  • Python 3.13
  • Role: GATT central, LE only

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions