trouble-host 0.8.0

//! BleHost
//!
//! The host module contains the main entry point for the TrouBLE host.
#[cfg(feature = "security")]
use core::cell::Cell;
use core::cell::RefCell;
use core::future::poll_fn;
use core::mem::MaybeUninit;
use core::task::{Context, Poll};

use bt_hci::cmd::controller_baseband::{
    HostBufferSize, HostNumberOfCompletedPackets, Reset, SetControllerToHostFlowControl, SetEventMask,
    SetEventMaskPage2,
};
use bt_hci::cmd::info::ReadBdAddr;
#[cfg(feature = "subrating")]
use bt_hci::cmd::le::LeSetHostFeature;
#[cfg(feature = "shorter-connection-intervals")]
use bt_hci::cmd::le::LeSetHostFeatureV2;
#[cfg(feature = "security")]
use bt_hci::cmd::le::{
    LeAddDeviceToResolvingList, LeClearResolvingList, LeRand, LeRemoveDeviceFromResolvingList,
    LeSetAddrResolutionEnable, LeSetPrivacyMode, LeSetResolvablePrivateAddrTimeout,
};
use bt_hci::cmd::le::{
    LeConnUpdate, LeCreateConnCancel, LeReadBufferSize, LeReadFilterAcceptListSize, LeSetAdvEnable, LeSetEventMask,
    LeSetExtAdvEnable, LeSetExtScanEnable, LeSetRandomAddr, LeSetScanEnable,
};
use bt_hci::cmd::link_control::Disconnect;
use bt_hci::cmd::{self, AsyncCmd, SyncCmd};
use bt_hci::controller::{blocking, Controller, ControllerCmdAsync, ControllerCmdSync};
#[cfg(feature = "iso")]
use bt_hci::data::IsoPacket;
use bt_hci::data::{AclBroadcastFlag, AclPacket, AclPacketBoundary};
#[cfg(feature = "scan")]
use bt_hci::event::le::LeAdvertisingReport;
#[cfg(feature = "scan")]
use bt_hci::event::le::LeExtendedAdvertisingReport;
#[cfg(feature = "subrating")]
use bt_hci::event::le::LeSubrateChange;
use bt_hci::event::le::{
    LeAdvertisingSetTerminated, LeConnectionComplete, LeConnectionRateChange, LeConnectionUpdateComplete,
    LeDataLengthChange, LeEnhancedConnectionComplete, LeEventKind, LeEventPacket, LeFrameSpaceUpdateComplete,
    LePhyUpdateComplete, LeRemoteConnectionParameterRequest,
};
#[cfg(feature = "iso")]
use bt_hci::event::le::{LeCisEstablished, LeCisRequest};
use bt_hci::event::{DisconnectionComplete, EventKind, NumberOfCompletedPackets, Vendor};
#[cfg(feature = "security")]
use bt_hci::param::BdAddr;
#[cfg(feature = "scan")]
use bt_hci::param::FilterDuplicates;
use bt_hci::param::{
    AddrKind, AdvHandle, AdvSet, ConnHandle, DisconnectReason, EventMask, EventMaskPage2, LeConnRole, LeEventMask,
    Status,
};
use bt_hci::{ControllerToHostPacket, FromHciBytes, WriteHci};
use embassy_futures::select::{select3, select5, Either3, Either5};
#[cfg(any(feature = "scan", feature = "security"))]
use embassy_sync::blocking_mutex::raw::NoopRawMutex;
#[cfg(feature = "security")]
use embassy_sync::mutex::Mutex;
use embassy_sync::once_lock::OnceLock;
#[cfg(feature = "scan")]
use embassy_sync::signal::Signal;
use embassy_sync::waitqueue::WakerRegistration;
use embassy_time::Duration;
#[cfg(all(feature = "security", feature = "central"))]
use embassy_time::{Instant, Timer};
use futures::pin_mut;

use crate::att::{AttClient, AttServer};
use crate::channel_manager::{ChannelManager, ChannelStorage};
use crate::command::CommandState;
use crate::connection::{ConnParams, ConnectionEvent};
#[cfg(feature = "security")]
use crate::connection_manager::ResolvablePrivateAddrs;
use crate::connection_manager::{AclSendLock, ConnectionManager, ConnectionStorage};
use crate::cursor::WriteCursor;
use crate::pdu::Pdu;
use crate::prelude::{ConnectionParamsRequest, RequestedConnParams};
#[cfg(feature = "security")]
use crate::security_manager::SecurityEventData;
use crate::types::l2cap::{
    ConnParamUpdateReq, ConnParamUpdateRes, L2capHeader, L2capSignal, L2capSignalHeader, L2CAP_CID_ATT,
    L2CAP_CID_DYN_START, L2CAP_CID_LE_U_SECURITY_MANAGER, L2CAP_CID_LE_U_SIGNAL,
};
use crate::{att, Address, BleHostError, Error, PacketPool};

#[cfg(feature = "security")]
#[derive(Clone, Copy)]
pub(crate) enum ResolvingListUpdate {
    FullSync,
    Add(crate::Identity),
    Remove(crate::Identity),
}

#[cfg(feature = "security")]
pub(crate) struct ResolvingListSignal {
    state: Option<ResolvingListUpdate>,
    waker: WakerRegistration,
}

#[cfg(feature = "security")]
impl ResolvingListSignal {
    const fn new() -> Self {
        Self {
            state: None,
            waker: WakerRegistration::new(),
        }
    }

    /// Push an update and wake the control loop.
    pub(crate) fn push(&mut self, update: ResolvingListUpdate) {
        self.state = Some(match self.state {
            Some(_) => ResolvingListUpdate::FullSync,
            None => update,
        });
        self.waker.wake();
    }

    /// Discard any pending update.
    pub(crate) fn clear(&mut self) {
        self.state = None;
    }

    /// Poll for a pending update. Takes the value on Ready.
    fn poll_changed(&mut self, cx: &mut Context<'_>) -> Poll<ResolvingListUpdate> {
        self.waker.register(cx.waker());
        match self.state.take() {
            Some(state) => Poll::Ready(state),
            None => Poll::Pending,
        }
    }
}

pub(crate) struct HostState<'d, P: PacketPool> {
    initialized: OnceLock<InitialState>,
    metrics: RefCell<HostMetrics>,
    pub(crate) address: Option<Address>,
    pub(crate) connections: ConnectionManager<'d, P>,
    pub(crate) channels: ChannelManager<'d, P>,
    pub(crate) advertise_state: AdvState<'d>,
    pub(crate) advertise_command_state: CommandState<bool>,
    pub(crate) connect_command_state: CommandState<bool>,
    pub(crate) scan_command_state: CommandState<bool>,
    #[cfg(feature = "security")]
    pub(crate) command_request_gate: Mutex<NoopRawMutex, ()>,
    #[cfg(feature = "security")]
    pub(crate) rpa_timeout: Cell<embassy_time::Duration>,
    #[cfg(all(feature = "security", feature = "central"))]
    pub(crate) rpa_expires_at: Cell<Instant>,
    #[cfg(feature = "security")]
    pub(crate) resolving_list_state: RefCell<ResolvingListSignal>,
    #[cfg(feature = "scan")]
    pub(crate) scan_timeout: Signal<NoopRawMutex, ()>,
}

impl<'d, P: PacketPool> HostState<'d, P> {
    pub(crate) fn new(
        connections: &'d RefCell<[ConnectionStorage<P::Packet>]>,
        channels: &'d RefCell<[ChannelStorage<P::Packet>]>,
        advertise_handles: &'d RefCell<[AdvHandleState]>,
        #[cfg(feature = "security")] bond_storage: &'d RefCell<heapless::VecView<crate::BondInformation>>,
    ) -> Self {
        Self {
            address: None,
            initialized: OnceLock::new(),
            metrics: RefCell::new(HostMetrics::default()),
            connections: ConnectionManager::new(
                connections,
                #[cfg(feature = "security")]
                bond_storage,
            ),
            channels: ChannelManager::new(channels),
            advertise_state: AdvState::new(advertise_handles),
            advertise_command_state: CommandState::new(),
            scan_command_state: CommandState::new(),
            connect_command_state: CommandState::new(),
            #[cfg(feature = "security")]
            command_request_gate: Mutex::new(()),
            #[cfg(feature = "security")]
            rpa_timeout: Cell::new(embassy_time::Duration::from_secs(900)),
            #[cfg(all(feature = "security", feature = "central"))]
            rpa_expires_at: Cell::new(Instant::from_ticks(0)),
            #[cfg(feature = "security")]
            resolving_list_state: RefCell::new(ResolvingListSignal::new()),
            #[cfg(feature = "scan")]
            scan_timeout: Signal::new(),
        }
    }
}

/// A BLE Host.
///
/// The BleHost holds the runtime state of the host, and is the entry point
/// for all interactions with the controller.
///
/// The host performs connection management, l2cap channel management, and
/// multiplexes events and data across connections and l2cap channels.
pub(crate) struct BleHost<'d, T, P: PacketPool> {
    controller: &'d T,
    state: &'d HostState<'d, P>,
}

impl<'d, T, P: PacketPool> Clone for BleHost<'d, T, P> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<'d, T, P: PacketPool> Copy for BleHost<'d, T, P> {}

#[derive(Clone, Copy)]
pub(crate) struct InitialState {
    acl_max: usize,
    acl_total: usize,
}

#[cfg_attr(feature = "defmt", derive(defmt::Format))]
#[derive(Clone, Copy, Debug)]
pub(crate) enum AdvHandleState {
    None,
    Advertising(AdvHandle),
    Terminated(AdvHandle),
}

pub(crate) struct AdvState<'d> {
    handles: &'d RefCell<[AdvHandleState]>,
    waker: RefCell<WakerRegistration>,
}

impl<'d> AdvState<'d> {
    pub(crate) fn new(handles: &'d RefCell<[AdvHandleState]>) -> Self {
        Self {
            handles,
            waker: RefCell::new(WakerRegistration::new()),
        }
    }

    pub(crate) fn reset(&self) {
        for entry in self.handles.borrow_mut().iter_mut() {
            *entry = AdvHandleState::None;
        }
        self.waker.borrow_mut().wake();
    }

    // Terminate handle
    pub(crate) fn terminate(&self, handle: AdvHandle) {
        for entry in self.handles.borrow_mut().iter_mut() {
            match entry {
                AdvHandleState::Advertising(h) if *h == handle => {
                    *entry = AdvHandleState::Terminated(handle);
                }
                _ => {}
            }
        }
        self.waker.borrow_mut().wake();
    }

    pub(crate) fn len(&self) -> usize {
        self.handles.as_ptr().len()
    }

    pub(crate) fn start(&self, sets: &[AdvSet]) {
        assert!(sets.len() <= self.handles.as_ptr().len());
        let mut handles = self.handles.borrow_mut();
        for handle in handles.iter_mut() {
            *handle = AdvHandleState::None;
        }

        for (idx, entry) in sets.iter().enumerate() {
            handles[idx] = AdvHandleState::Advertising(entry.adv_handle);
        }
    }

    pub async fn wait(&self) {
        poll_fn(|cx| {
            self.waker.borrow_mut().register(cx.waker());

            let mut terminated = 0;
            for entry in self.handles.borrow().iter() {
                match entry {
                    AdvHandleState::Terminated(_) => {
                        terminated += 1;
                    }
                    AdvHandleState::None => {
                        terminated += 1;
                    }
                    _ => {}
                }
            }
            if terminated == self.handles.as_ptr().len() {
                Poll::Ready(())
            } else {
                Poll::Pending
            }
        })
        .await;
    }
}

/// Host metrics
#[derive(Default, Clone)]
pub struct HostMetrics {
    /// How many connect events have been received.
    pub connect_events: u32,
    /// How many disconnect events have been received.
    pub disconnect_events: u32,
    /// How many errors processing received data.
    pub rx_errors: u32,
}

impl<'d, T, P> BleHost<'d, T, P>
where
    P: PacketPool,
{
    /// Create a new instance of the BLE host.
    ///
    /// The host requires a HCI driver (a particular HCI-compatible controller implementing the required traits), and
    /// a reference to resources that are created outside the host but which the host is the only accessor of.
    pub(crate) fn new(controller: &'d T, state: &'d HostState<'d, P>) -> Self {
        Self { controller, state }
    }

    pub(crate) fn address(&self) -> Option<Address> {
        self.state.address
    }

    pub(crate) fn connect_command_state(&self) -> &CommandState<bool> {
        &self.state.connect_command_state
    }

    pub(crate) fn advertise_command_state(&self) -> &CommandState<bool> {
        &self.state.advertise_command_state
    }

    pub(crate) fn scan_command_state(&self) -> &'d CommandState<bool> {
        &self.state.scan_command_state
    }

    pub(crate) fn connections(&self) -> &'d ConnectionManager<'d, P> {
        &self.state.connections
    }

    pub(crate) fn channels(&self) -> &'d ChannelManager<'d, P> {
        &self.state.channels
    }

    pub(crate) fn advertise_state(&self) -> &AdvState<'d> {
        &self.state.advertise_state
    }

    #[cfg(feature = "scan")]
    pub(crate) fn scan_timeout(&self) -> &'d Signal<NoopRawMutex, ()> {
        &self.state.scan_timeout
    }

    #[cfg(feature = "security")]
    pub(crate) fn resolving_list_state(&self) -> &RefCell<ResolvingListSignal> {
        &self.state.resolving_list_state
    }

    #[cfg(feature = "security")]
    pub(crate) fn rpa_timeout(&self) -> &Cell<Duration> {
        &self.state.rpa_timeout
    }
}

impl<'d, T, P> BleHost<'d, T, P>
where
    T: Controller,
    P: PacketPool,
{
    /// Poll whether any command should be cancelled or the resolving list should be synced.
    fn poll_cancelled(&self, cx: &mut Context<'_>) -> Poll<CancelledCommandState> {
        // Not every branch below survives every feature combination.
        let _ = cx;

        #[cfg(feature = "central")]
        if let Poll::Ready(ctx) = self.state.connect_command_state.poll_cancelled(cx) {
            return Poll::Ready(CancelledCommandState::Connect(ctx));
        }
        #[cfg(feature = "peripheral")]
        if let Poll::Ready(ctx) = self.state.advertise_command_state.poll_cancelled(cx) {
            return Poll::Ready(CancelledCommandState::Advertise(ctx));
        }
        #[cfg(feature = "scan")]
        if let Poll::Ready(ctx) = self.state.scan_command_state.poll_cancelled(cx) {
            return Poll::Ready(CancelledCommandState::Scan(ctx));
        }

        #[cfg(all(feature = "security", feature = "central"))]
        if self.is_privacy_enabled() && self.is_rpa_rotation_ready() {
            return Poll::Ready(CancelledCommandState::RotateRpa);
        }

        #[cfg(feature = "security")]
        if self.state.connect_command_state.is_idle()
            && self.state.advertise_command_state.is_idle()
            && self.state.scan_command_state.is_idle()
        {
            if let Poll::Ready(update) = self.state.resolving_list_state.borrow_mut().poll_changed(cx) {
                return Poll::Ready(CancelledCommandState::SyncResolvingList(update));
            }
        }

        Poll::Pending
    }

    /// Check whether BLE address privacy is enabled.
    #[cfg(feature = "security")]
    pub(crate) fn is_privacy_enabled(&self) -> bool {
        self.state.connections.security_manager.get_local_irk().is_some()
    }

    /// Get the appropriate own address kind based on the host address and privacy state.
    pub(crate) fn own_addr_kind(&self) -> AddrKind {
        #[cfg(feature = "security")]
        if self.is_privacy_enabled() {
            return if self.state.address.is_some() {
                AddrKind::RESOLVABLE_PRIVATE_OR_RANDOM
            } else {
                AddrKind::RESOLVABLE_PRIVATE_OR_PUBLIC
            };
        }
        self.state.address.map(|a| a.kind).unwrap_or(AddrKind::PUBLIC)
    }

    /// Atomically mark an address-using procedure active relative to RPA rotation.
    pub(crate) async fn request_operation<CTX: Clone + Copy>(&self, state: &CommandState<CTX>, ctx: CTX) {
        #[cfg(feature = "security")]
        let _guard = self.state.command_request_gate.lock().await;
        state.request(ctx).await;
    }

    #[cfg(all(feature = "security", feature = "central"))]
    fn is_rpa_rotation_ready(&self) -> bool {
        self.is_privacy_enabled()
            && self.state.connect_command_state.is_idle()
            && !self.state.advertise_command_state.is_active_with(|extended| !extended)
            && self.state.scan_command_state.is_idle()
            && Instant::now() >= self.state.rpa_expires_at.get()
    }

    #[cfg(all(feature = "security", feature = "central"))]
    async fn wait_for_rpa_expiration(&self) {
        while Instant::now() < self.state.rpa_expires_at.get() {
            Timer::at(self.state.rpa_expires_at.get()).await
        }
        if !self.is_rpa_rotation_ready() {
            // A command is still active, so rely on poll_cancelled to wake the runner loop instead
            core::future::pending::<()>().await;
        }
    }

    #[cfg(all(feature = "security", feature = "central"))]
    async fn rotate_rpa(&self) -> Result<(), BleHostError<T::Error>>
    where
        T: ControllerCmdSync<LeSetRandomAddr>,
    {
        let _guard = self.state.command_request_gate.lock().await;
        if !self.is_rpa_rotation_ready() {
            return Ok(());
        }
        let Some(rpa) = self.state.connections.security_manager.generate_local_rpa() else {
            return Ok(());
        };
        LeSetRandomAddr::new(rpa).exec(self.controller).await?;
        self.state
            .rpa_expires_at
            .set(Instant::now() + self.state.rpa_timeout.get());
        trace!("[host] rotated private address");
        Ok(())
    }

    /// Sync the controller's resolving list based on a pending update.
    #[cfg(feature = "security")]
    pub(crate) async fn sync_resolving_list(&self, update: ResolvingListUpdate) -> Result<(), BleHostError<T::Error>>
    where
        T: ControllerCmdSync<LeClearResolvingList>
            + ControllerCmdSync<LeSetAddrResolutionEnable>
            + ControllerCmdSync<LeRemoveDeviceFromResolvingList>
            + ControllerCmdSync<LeAddDeviceToResolvingList>
            + ControllerCmdSync<LeSetPrivacyMode>,
        T::Error: crate::fmt::Format,
    {
        let _guard = self.state.command_request_gate.lock().await;
        if !(self.state.connect_command_state.is_idle()
            && self.state.advertise_command_state.is_idle()
            && self.state.scan_command_state.is_idle())
        {
            return Ok(());
        }

        let local_irk = self.state.connections.security_manager.get_local_irk();
        let local_irk_bytes = local_irk.map(|k| k.to_le_bytes()).unwrap_or_default();

        // Disable address resolution while modifying the list
        LeSetAddrResolutionEnable::new(false).exec(self.controller).await?;

        let res = match update {
            ResolvingListUpdate::FullSync => self.full_resolving_list_sync(local_irk).await,
            ResolvingListUpdate::Add(identity) => {
                if let Some(peer_irk) = identity.irk {
                    let peer_addr_kind = identity.addr.kind;
                    let peer_irk_bytes = peer_irk.to_le_bytes();

                    debug!("[host] incremental resolving list add");

                    // Remove first in case this is an update (device already in list)
                    let _ = LeRemoveDeviceFromResolvingList::new(peer_addr_kind, identity.addr.addr)
                        .exec(self.controller)
                        .await;

                    if let Err(e) = LeAddDeviceToResolvingList::new(
                        peer_addr_kind,
                        identity.addr.addr,
                        peer_irk_bytes,
                        local_irk_bytes,
                    )
                    .exec(self.controller)
                    .await
                    {
                        warn!("[host] failed to add device to resolving list: {:?}", e);
                    }

                    if let Err(e) =
                        LeSetPrivacyMode::new(peer_addr_kind, identity.addr.addr, bt_hci::param::PrivacyMode::Device)
                            .exec(self.controller)
                            .await
                    {
                        warn!("[host] failed to set privacy mode: {:?}", e);
                    }
                }
                Ok(())
            }
            ResolvingListUpdate::Remove(identity) => {
                let peer_addr_kind = identity.addr.kind;

                debug!("[host] incremental resolving list remove");

                if let Err(e) = LeRemoveDeviceFromResolvingList::new(peer_addr_kind, identity.addr.addr)
                    .exec(self.controller)
                    .await
                {
                    warn!("[host] failed to remove device from resolving list: {:?}", e);
                }
                Ok(())
            }
        };

        // Always re-enable address resolution
        if let Err(e) = LeSetAddrResolutionEnable::new(true).exec(self.controller).await {
            warn!("[host] failed to re-enable address resolution: {:?}", e);
            return res.and(Err(e.into()));
        }

        res
    }

    /// Full clear-and-rebuild of the resolving list.
    #[cfg(feature = "security")]
    async fn full_resolving_list_sync(
        &self,
        local_irk: Option<crate::security_manager::IdentityResolvingKey>,
    ) -> Result<(), BleHostError<T::Error>>
    where
        T: ControllerCmdSync<LeClearResolvingList>
            + ControllerCmdSync<LeAddDeviceToResolvingList>
            + ControllerCmdSync<LeSetPrivacyMode>,
        T::Error: crate::fmt::Format,
    {
        debug!("[host] full resolving list sync");

        // Clear resolving list
        LeClearResolvingList::new().exec(self.controller).await?;

        let local_irk_bytes = local_irk.map(|k| k.to_le_bytes()).unwrap_or_default();

        // Add a default entry with just the local IRK so the controller can generate RPAs
        // even when no peers are bonded yet. Only needed when we have our own IRK.
        if local_irk.is_some() {
            if let Err(e) =
                LeAddDeviceToResolvingList::new(AddrKind::PUBLIC, BdAddr::default(), [0u8; 16], local_irk_bytes)
                    .exec(self.controller)
                    .await
            {
                warn!("[host] failed to add default resolving list entry: {:?}", e);
            }
        }

        // Add entries for all bonded peers with IRKs
        let mut i = 0;
        while let Some(bond) = self.state.connections.security_manager.get_bond(i) {
            i += 1;
            if let Some(peer_irk) = bond.identity.irk {
                let peer_addr_kind = bond.identity.addr.kind;
                let peer_irk_bytes = peer_irk.to_le_bytes();
                if let Err(e) = LeAddDeviceToResolvingList::new(
                    peer_addr_kind,
                    bond.identity.addr.addr,
                    peer_irk_bytes,
                    local_irk_bytes,
                )
                .exec(self.controller)
                .await
                {
                    warn!("[host] failed to add device to resolving list: {:?}", e);
                    break;
                }
                // Set device privacy mode so we can receive both RPAs and identity addresses
                if let Err(e) = LeSetPrivacyMode::new(
                    peer_addr_kind,
                    bond.identity.addr.addr,
                    bt_hci::param::PrivacyMode::Device,
                )
                .exec(self.controller)
                .await
                {
                    warn!("[host] failed to set privacy mode: {:?}", e);
                }
            }
        }

        debug!("[host] resolving list synced");
        Ok(())
    }

    /// Returns true if the host has been initialized by the Runner.
    pub(crate) fn is_initialized(&self) -> bool {
        self.state.initialized.try_get().is_some()
    }

    /// Run a HCI command and return the response.
    pub(crate) async fn command<C>(&self, cmd: C) -> Result<C::Return, BleHostError<T::Error>>
    where
        C: SyncCmd,
        T: ControllerCmdSync<C>,
    {
        let _ = self.state.initialized.get().await;
        let ret = cmd.exec(self.controller).await?;
        Ok(ret)
    }

    /// Run an async HCI command where the response will generate an event later.
    pub(crate) async fn async_command<C>(&self, cmd: C) -> Result<(), BleHostError<T::Error>>
    where
        C: AsyncCmd,
        T: ControllerCmdAsync<C>,
    {
        let _ = self.state.initialized.get().await;
        cmd.exec(self.controller).await?;
        Ok(())
    }

    #[cfg(feature = "iso")]
    pub(crate) async fn write_iso_data(&self, packet: &bt_hci::data::IsoPacket<'_>) -> Result<(), T::Error> {
        self.controller.write_iso_data(packet).await
    }

    fn handle_connection(
        &self,
        status: Status,
        handle: ConnHandle,
        peer_addr: Address,
        role: LeConnRole,
        params: ConnParams,
        #[cfg(feature = "security")] resolvable_addrs: ResolvablePrivateAddrs,
    ) -> bool {
        match status.to_result() {
            Ok(_) => {
                if let Err(err) = self.state.connections.connect_with_rpas(
                    handle,
                    peer_addr,
                    role,
                    params,
                    #[cfg(feature = "security")]
                    resolvable_addrs,
                ) {
                    warn!("Error establishing connection: {:?}", err);
                    return false;
                } else {
                    #[cfg(feature = "defmt")]
                    debug!(
                        "[host] connection with handle {:?} established to {}",
                        handle, peer_addr
                    );

                    #[cfg(feature = "log")]
                    debug!(
                        "[host] connection with handle {:?} established to {}",
                        handle, peer_addr
                    );
                    let mut m = self.state.metrics.borrow_mut();
                    m.connect_events = m.connect_events.wrapping_add(1);
                }
            }
            Err(bt_hci::param::Error::ADV_TIMEOUT) => {
                self.state.advertise_state.reset();
            }
            Err(bt_hci::param::Error::UNKNOWN_CONN_IDENTIFIER) => {
                warn!("[host] connect cancelled");
                self.state.connect_command_state.canceled();
            }
            Err(e) => {
                warn!("Error connection complete event: {:?}", e);
                self.state.connect_command_state.canceled();
            }
        }
        true
    }

    fn handle_acl(&self, acl: AclPacket<'_>, event_handler: &dyn EventHandler) -> Result<(), Error> {
        self.state.connections.received(acl.handle())?;
        let handle = acl.handle();
        let (header, pdu) = match acl.boundary_flag() {
            AclPacketBoundary::FirstFlushable => {
                let (header, data) = L2capHeader::from_hci_bytes(acl.data())?;

                // Ignore channels we don't support
                if header.channel < L2CAP_CID_DYN_START
                    && !(&[L2CAP_CID_LE_U_SIGNAL, L2CAP_CID_ATT, L2CAP_CID_LE_U_SECURITY_MANAGER]
                        .contains(&header.channel))
                {
                    // Apple devices probe this channel id even if it's outside the dynamic range.
                    if header.channel == 0x3a {
                        info!(
                            "[host] unsupported l2cap channel id {} (Apple devices which always probe this channel, so this is safe to ignore).",
                            header.channel
                        );
                    } else {
                        warn!("[host] unsupported l2cap channel id {}", header.channel);
                    }
                    return Err(Error::NotSupported);
                }

                // Fast-path for complete signalling packets
                if header.channel == L2CAP_CID_LE_U_SIGNAL && data.len() == header.length as usize {
                    return self.state.channels.signal(acl.handle(), data, &self.state.connections);
                }

                trace!(
                    "[host] inbound l2cap header channel = {}, fragment len = {}, total = {}",
                    header.channel,
                    data.len(),
                    header.length
                );

                // We must be prepared to receive fragments.
                if header.length as usize != data.len() {
                    // Dynamic channels can be optimized.
                    #[cfg(feature = "l2cap-sdu-reassembly-optimization")]
                    if header.channel >= L2CAP_CID_DYN_START {
                        // This is the start of the frame, so make sure to adjust the credits.
                        self.state.channels.received(header.channel, 1)?;
                        self.state.channels.check_pdu_len(header.channel, header.length)?;

                        self.state
                            .connections
                            .reassembly(acl.handle(), |p| {
                                let r = if !p.in_progress() {
                                    // Init the new assembly assuming the length of the SDU.
                                    let (first, payload) = data.split_at(2);
                                    let len: u16 = u16::from_le_bytes([first[0], first[1]]);
                                    self.state.channels.check_sdu_len(header.channel, len)?;
                                    let Some(packet) = P::allocate() else {
                                        warn!("[host] no memory for packets on channel {}", header.channel);
                                        return Err(Error::OutOfMemory);
                                    };
                                    p.init(header.channel, len, packet)?;
                                    p.update(payload)?
                                } else {
                                    p.update(data)?
                                };
                                // Something is wrong if assembly was finished since we've not received the last fragment.
                                if r.is_some() {
                                    Err(Error::InvalidState)
                                } else {
                                    Ok(())
                                }
                            })
                            .inspect_err(|_| self.state.channels.disconnect_by_cid(header.channel))?;
                        return Ok(());
                    }

                    // For dynamic channels, validate PDU length against MPS before reassembly.
                    if header.channel >= L2CAP_CID_DYN_START {
                        self.state.channels.check_pdu_len(header.channel, header.length)?;
                    }

                    let Some(packet) = P::allocate() else {
                        warn!("[host] no memory for packets on channel {}", header.channel);
                        return Err(Error::OutOfMemory);
                    };
                    self.state.connections.reassembly(acl.handle(), |p| {
                        p.init(header.channel, header.length, packet)?;
                        let r = p.update(data)?;
                        if r.is_some() {
                            Err(Error::InvalidState)
                        } else {
                            Ok(())
                        }
                    })?;
                    return Ok(());
                } else {
                    #[allow(unused_mut)]
                    let mut result = None;

                    // For dynamic channels, validate PDU length against MPS and handle
                    // credit accounting and SDU reassembly before HCI reassembly.
                    #[cfg(feature = "l2cap-sdu-reassembly-optimization")]
                    if header.channel >= L2CAP_CID_DYN_START {
                        // This is a complete L2CAP K-frame, so make sure to adjust the credits.
                        self.state.channels.received(header.channel, 1)?;
                        self.state.channels.check_pdu_len(header.channel, header.length)?;

                        if let Some((state, pdu)) = self
                            .state
                            .connections
                            .reassembly(acl.handle(), |p| {
                                if !p.in_progress() {
                                    let (first, payload) = data.split_at(2);
                                    let len: u16 = u16::from_le_bytes([first[0], first[1]]);
                                    self.state.channels.check_sdu_len(header.channel, len)?;
                                    let Some(packet) = P::allocate() else {
                                        warn!("[host] no memory for packets on channel {}", header.channel);
                                        return Err(Error::OutOfMemory);
                                    };
                                    p.init(header.channel, len, packet)?;
                                    p.update(payload)
                                } else {
                                    p.update(data)
                                }
                            })
                            .inspect_err(|_| self.state.channels.disconnect_by_cid(header.channel))?
                        {
                            result.replace((state, pdu));
                        } else {
                            return Ok(());
                        }
                    }

                    if let Some((state, pdu)) = result {
                        (state, pdu)
                    } else {
                        // For dynamic channels, validate PDU length against MPS before reassembly.
                        if header.channel >= L2CAP_CID_DYN_START {
                            self.state.channels.check_pdu_len(header.channel, header.length)?;
                        }

                        let Some(packet) = P::allocate() else {
                            warn!("[host] no memory for packets on channel {}", header.channel);
                            return Err(Error::OutOfMemory);
                        };
                        let result = self.state.connections.reassembly(acl.handle(), |p| {
                            p.init(header.channel, header.length, packet)?;
                            p.update(data)
                        })?;
                        let Some((state, pdu)) = result else {
                            return Err(Error::InvalidState);
                        };
                        (state, pdu)
                    }
                }
            }
            // Next (potentially last) in a fragment
            AclPacketBoundary::Continuing => {
                trace!("[host] inbound l2cap len = {}", acl.data().len(),);
                // Get the existing fragment
                if let Some((header, p)) = self.state.connections.reassembly(acl.handle(), |p| {
                    if !p.in_progress() {
                        warn!(
                            "[host] unexpected continuation fragment of length {} for handle {}: {:?}",
                            acl.data().len(),
                            acl.handle().raw(),
                            p
                        );
                        return Err(Error::InvalidState);
                    }
                    p.update(acl.data())
                })? {
                    (header, p)
                } else {
                    // Do not process yet
                    return Ok(());
                }
            }
            other => {
                warn!("Unexpected boundary flag: {:?}!", other);
                return Err(Error::NotSupported);
            }
        };

        match header.channel {
            L2CAP_CID_ATT => {
                // Handle ATT MTU exchange here since it doesn't strictly require
                // gatt to be enabled.
                let a = att::Att::decode(pdu.as_ref());
                if let Ok(att::Att::Client(AttClient::Request(att::AttReq::ExchangeMtu { mtu }))) = a {
                    let mtu = self.state.connections.exchange_att_mtu(acl.handle(), mtu);

                    let rsp = att::Att::Server(AttServer::Response(att::AttRsp::ExchangeMtu { mtu }));
                    let l2cap = L2capHeader {
                        channel: L2CAP_CID_ATT,
                        length: 3,
                    };

                    let mut packet = pdu.into_inner();
                    let mut w = WriteCursor::new(packet.as_mut());
                    w.write_hci(&l2cap)?;
                    w.write(rsp)?;

                    debug!("[host] agreed att MTU of {}", mtu);
                    let len = w.len();
                    self.state
                        .connections
                        .try_outbound(acl.handle(), Pdu::new(packet, len))?;
                } else if let Ok(att::Att::Server(AttServer::Response(att::AttRsp::ExchangeMtu { mtu }))) = a {
                    debug!("[host] remote agreed att MTU of {}", mtu);
                    self.state.connections.exchange_att_mtu(acl.handle(), mtu);
                    #[cfg(feature = "gatt")]
                    self.state.connections.post_gatt_client(acl.handle(), pdu)?;
                } else {
                    #[cfg(feature = "gatt")]
                    match a {
                        Ok(att::Att::Client(AttClient::Confirmation(_))) => {
                            self.state.connections.signal_indication_confirmation(acl.handle());
                        }
                        Ok(att::Att::Client(_)) => {
                            self.state.connections.post_gatt(acl.handle(), pdu)?;
                        }
                        Ok(att::Att::Server(AttServer::Unsolicited(att::AttUns::Indicate { .. }))) => {
                            // Per BLE spec, ATT_HANDLE_VALUE_CFM must always be sent in response
                            // to an indication, regardless of whether a GATT client task is running.
                            let cfm = att::Att::Client(AttClient::Confirmation(crate::att::AttCfm::ConfirmIndication));
                            let l2cap = L2capHeader {
                                channel: L2CAP_CID_ATT,
                                length: cfm.size() as u16,
                            };
                            let mut buf = P::allocate().ok_or(Error::OutOfMemory)?;
                            let mut w = WriteCursor::new(buf.as_mut());
                            w.write_hci(&l2cap)?;
                            w.write(cfm)?;
                            let len = w.len();
                            self.state.connections.try_outbound(acl.handle(), Pdu::new(buf, len))?;

                            // Also queue the indication for the GATT client if possible
                            let _ = self.state.connections.post_gatt_client(acl.handle(), pdu);
                        }
                        Ok(att::Att::Server(_)) => {
                            if let Err(e) = self.state.connections.post_gatt_client(acl.handle(), pdu) {
                                return Err(Error::OutOfMemory);
                            }
                        }
                        Err(e) => {
                            warn!("Error decoding attribute payload: {:?}", e);
                            let opcode = pdu.as_ref()[0];
                            // Bit 6 = Command Flag. Only send error responses for requests (flag=0)
                            if opcode & 0x40 == 0 {
                                let rsp = att::Att::Server(AttServer::Response(att::AttRsp::Error {
                                    request: opcode,
                                    handle: 0,
                                    code: att::AttErrorCode::REQUEST_NOT_SUPPORTED,
                                }));
                                let l2cap = L2capHeader {
                                    channel: L2CAP_CID_ATT,
                                    length: rsp.size() as u16,
                                };
                                let mut packet = pdu.into_inner();
                                let mut w = WriteCursor::new(packet.as_mut());
                                w.write_hci(&l2cap)?;
                                w.write(rsp)?;
                                let len = w.len();
                                self.state
                                    .connections
                                    .try_outbound(acl.handle(), Pdu::new(packet, len))?;
                            }
                        }
                    }
                    #[cfg(not(feature = "gatt"))]
                    {
                        if let Ok(att::Att::Client(_)) = a {
                            drop(a);

                            let opcode = pdu.as_ref()[0];
                            let rsp = att::Att::Server(AttServer::Response(att::AttRsp::Error {
                                request: opcode,
                                handle: acl.handle().raw(),
                                code: att::AttErrorCode::ATTRIBUTE_NOT_FOUND,
                            }));

                            let mut packet = pdu.into_inner();
                            let mut w = WriteCursor::new(packet.as_mut());

                            let l2cap = L2capHeader {
                                channel: L2CAP_CID_ATT,
                                length: rsp.size() as u16,
                            };

                            w.write_hci(&l2cap)?;
                            w.write(rsp)?;

                            let len = w.len();
                            self.state
                                .connections
                                .try_outbound(acl.handle(), Pdu::new(packet, len))?;
                            warn!("[host] got attribute request but 'gatt' feature is not enabled.");
                            return Ok(());
                        } else {
                            warn!("Got unsupported ATT: {:?}", a);
                            return Err(Error::NotSupported);
                        }
                    }
                }
            }
            L2CAP_CID_LE_U_SIGNAL => {
                self.state
                    .channels
                    .signal(handle, pdu.as_ref(), &self.state.connections)?;
            }
            L2CAP_CID_LE_U_SECURITY_MANAGER => {
                self.state
                    .connections
                    .handle_security_channel(acl.handle(), pdu, event_handler)?;
            }
            other if other >= L2CAP_CID_DYN_START => match self.state.channels.dispatch(header.channel, pdu) {
                Ok(_) => {}
                Err(e) => {
                    warn!("Error dispatching l2cap packet to channel: {:?}", e);
                    return Err(e);
                }
            },
            chan => {
                debug!(
                    "[host] conn {:?} attempted to use unsupported l2cap channel {}, ignoring",
                    acl.handle(),
                    chan
                );
                return Ok(());
            }
        }
        Ok(())
    }

    // Send l2cap signal payload
    pub(crate) async fn l2cap_signal<D: L2capSignal>(
        &self,
        conn: ConnHandle,
        identifier: u8,
        signal: &D,
        p_buf: &mut [u8],
    ) -> Result<(), BleHostError<T::Error>> {
        //trace!(
        //    "[l2cap] sending control signal (req = {}) signal: {:?}",
        //    identifier,
        //    signal
        //);
        let header = L2capSignalHeader {
            identifier,
            code: D::code(),
            length: signal.size() as u16,
        };
        let l2cap = L2capHeader {
            channel: D::channel(),
            length: header.size() as u16 + header.length,
        };

        let mut w = WriteCursor::new(p_buf);
        w.write_hci(&l2cap)?;
        w.write_hci(&header)?;
        w.write_hci(signal)?;

        let mut sender = self.l2cap_pdu(conn).await?;
        sender.send(w.finish()).await?;

        Ok(())
    }

    // Request to send a single L2CAP PDU of pdu_len to the HCI controller for a connection.
    //
    // This function will acquire the connection ACL send lock and the returned sender will handle ACL fragmentation.
    //
    // This function cannot be used to send an SDU split among multiple k-frames.
    pub(crate) async fn l2cap_pdu(&self, handle: ConnHandle) -> Result<L2capSender<'_, T, P>, BleHostError<T::Error>> {
        // Take into account l2cap header.
        let initial_state = self.state.initialized.get().await;
        let acl_max = initial_state.acl_max as u16;
        if acl_max == 0 {
            return Err(Error::NoPermits.into());
        }

        let acl_send_lock = poll_fn(|cx| self.state.connections.poll_acquire_acl_send_lock(handle, Some(cx))).await?;
        Ok(L2capSender {
            controller: self.controller,
            handle,
            acl_send_lock,
            fragment_size: acl_max,
            max_fragments: initial_state.acl_total,
        })
    }

    // Request to send an L2CAP SDU of sdu_len to the HCI controller for a connection.
    //
    // This function will acquire the connection ACL send lock and the returned sender will handle ACL fragmentation.
    pub(crate) fn try_l2cap_sdu(
        &self,
        handle: ConnHandle,
        sdu_len: u16,
        mps: u16,
    ) -> Result<L2capSender<'_, T, P>, BleHostError<T::Error>> {
        let initial_state = self.state.initialized.try_get().ok_or(Error::NoPermits)?;
        let acl_max = initial_state.acl_max;
        if acl_max == 0 {
            return Err(Error::NoPermits.into());
        }

        let acl_send_lock = match self.state.connections.poll_acquire_acl_send_lock(handle, None) {
            Poll::Ready(res) => res?,
            Poll::Pending => {
                return Err(Error::Busy.into());
            }
        };

        const L2CAP_BASIC_HEADER_SIZE: usize = 4; // L2CAP Basic Header added to each k-frame
        const L2CAP_SDU_LEN_SIZE: usize = 2; // L2CAP SDU Length field added to first k-frame

        let mps = usize::from(mps);
        let len = usize::from(sdu_len) + L2CAP_SDU_LEN_SIZE;
        let full_k_frame_len = mps + L2CAP_BASIC_HEADER_SIZE;
        let full_k_frames = len / mps;
        let last_k_frame_len = if len.is_multiple_of(mps) {
            0
        } else {
            len % mps + L2CAP_BASIC_HEADER_SIZE
        };

        let n_acl = full_k_frames * full_k_frame_len.div_ceil(acl_max) + last_k_frame_len.div_ceil(acl_max);
        if n_acl > initial_state.acl_total {
            return Err(Error::NoPermits.into());
        } else if !acl_send_lock.has_link_credits(n_acl) {
            return Err(Error::Busy.into());
        }

        Ok(L2capSender {
            controller: self.controller,
            handle,
            acl_send_lock,
            fragment_size: acl_max as u16,
            max_fragments: initial_state.acl_total,
        })
    }

    pub(crate) async fn send_conn_param_update_req(
        &self,
        handle: ConnHandle,
        param: &ConnParamUpdateReq,
    ) -> Result<(), BleHostError<T::Error>> {
        self.state
            .channels
            .send_conn_param_update_req(handle, self, param)
            .await
    }

    pub(crate) async fn send_conn_param_update_res(
        &self,
        handle: ConnHandle,
        param: &ConnParamUpdateRes,
    ) -> Result<(), BleHostError<T::Error>> {
        self.state
            .channels
            .send_conn_param_update_res(handle, self, param)
            .await
    }

    /// Read current host metrics
    pub(crate) fn metrics<F: FnOnce(&HostMetrics) -> R, R>(&self, f: F) -> R {
        let m = self.state.metrics.borrow();
        f(&m)
    }

    /// Log status information of the host
    pub(crate) fn log_status(&self, verbose: bool) {
        let m = self.state.metrics.borrow();
        debug!("[host] connect events: {}", m.connect_events);
        debug!("[host] disconnect events: {}", m.disconnect_events);
        debug!("[host] rx errors: {}", m.rx_errors);
        self.state.connections.log_status(verbose);
        self.state.channels.log_status(verbose);
    }
}

/// Runs the host with the given controller.
pub struct Runner<'d, C, P: PacketPool> {
    rx: RxRunner<'d, C, P>,
    control: ControlRunner<'d, C, P>,
    tx: TxRunner<'d, C, P>,
}

/// The receiver part of the host runner.
pub struct RxRunner<'d, C, P: PacketPool> {
    host: BleHost<'d, C, P>,
}

/// The control part of the host runner.
pub struct ControlRunner<'d, C, P: PacketPool> {
    host: BleHost<'d, C, P>,
}

/// The transmit part of the host runner.
pub struct TxRunner<'d, C, P: PacketPool> {
    host: BleHost<'d, C, P>,
}

/// Event handler.
pub trait EventHandler {
    /// Handle vendor events
    fn on_vendor(&self, vendor: &Vendor) {}
    /// Handle advertising reports
    #[cfg(feature = "scan")]
    fn on_adv_reports(&self, reports: bt_hci::param::LeAdvReportsIter) {}
    /// Handle extended advertising reports
    #[cfg(feature = "scan")]
    fn on_ext_adv_reports(&self, reports: bt_hci::param::LeExtAdvReportsIter) {}
    /// Handle HCI NumberOfCompletedPackets event.
    ///
    /// Called for each connection handle entry when the controller confirms
    /// ACL data packets have been transmitted over the air. Useful for
    /// measuring actual air delivery rate and estimating connection event timing.
    fn on_packets_completed(&self, _num_completed: usize) {}

    /// Handle an LE CIS Request event
    #[cfg(feature = "iso")]
    fn on_cis_request(&self, _event: &LeCisRequest) {}
    /// Handle an LE CIS Established event
    #[cfg(feature = "iso")]
    fn on_cis_established(&self, _event: &LeCisEstablished) {}
    /// Handle an incoming HCI ISO data packet
    #[cfg(feature = "iso")]
    fn on_iso_data(&self, _packet: &IsoPacket<'_>) {}
}

struct DummyHandler;
impl EventHandler for DummyHandler {}

impl<'d, C: Controller, P: PacketPool> Runner<'d, C, P> {
    pub(crate) fn new(host: BleHost<'d, C, P>) -> Self {
        Self {
            rx: RxRunner { host },
            control: ControlRunner { host },
            tx: TxRunner { host },
        }
    }

    /// Split the runner into separate independent async tasks
    pub fn split(self) -> (RxRunner<'d, C, P>, ControlRunner<'d, C, P>, TxRunner<'d, C, P>) {
        (self.rx, self.control, self.tx)
    }

    /// Run the host.
    pub async fn run(&mut self) -> Result<(), BleHostError<C::Error>>
    where
        C: ControllerCmdSync<Disconnect>
            + ControllerCmdSync<SetEventMask>
            + ControllerCmdSync<SetEventMaskPage2>
            + ControllerCmdSync<LeSetEventMask>
            + ControllerCmdSync<LeSetRandomAddr>
            + ControllerCmdSync<HostBufferSize>
            + ControllerCmdAsync<LeConnUpdate>
            + ControllerCmdSync<LeReadFilterAcceptListSize>
            + ControllerCmdSync<SetControllerToHostFlowControl>
            + ControllerCmdSync<Reset>
            + ControllerCmdSync<LeCreateConnCancel>
            + ControllerCmdSync<LeSetScanEnable>
            + ControllerCmdSync<LeSetExtScanEnable>
            + for<'t> ControllerCmdSync<LeSetAdvEnable>
            + for<'t> ControllerCmdSync<LeSetExtAdvEnable<'t>>
            + for<'t> ControllerCmdSync<HostNumberOfCompletedPackets<'t>>
            + ControllerCmdSync<LeReadBufferSize>
            + ControllerCmdSync<ReadBdAddr>
            + crate::SecurityCmds
            + crate::IsoStreamCmds
            + crate::SubratingCmds
            + crate::ShortConnIntervalCmds,
        C::Error: crate::fmt::Format,
    {
        let dummy = DummyHandler;
        self.run_with_handler(&dummy).await
    }

    /// Run the host with a vendor event handler for custom events.
    pub async fn run_with_handler<E: EventHandler>(&mut self, event_handler: &E) -> Result<(), BleHostError<C::Error>>
    where
        C: ControllerCmdSync<Disconnect>
            + ControllerCmdSync<SetEventMask>
            + ControllerCmdSync<SetEventMaskPage2>
            + ControllerCmdSync<LeSetEventMask>
            + ControllerCmdSync<LeSetRandomAddr>
            + ControllerCmdSync<LeReadFilterAcceptListSize>
            + ControllerCmdSync<HostBufferSize>
            + ControllerCmdAsync<LeConnUpdate>
            + ControllerCmdSync<SetControllerToHostFlowControl>
            + for<'t> ControllerCmdSync<LeSetAdvEnable>
            + for<'t> ControllerCmdSync<LeSetExtAdvEnable<'t>>
            + for<'t> ControllerCmdSync<HostNumberOfCompletedPackets<'t>>
            + ControllerCmdSync<LeSetScanEnable>
            + ControllerCmdSync<LeSetExtScanEnable>
            + ControllerCmdSync<Reset>
            + ControllerCmdSync<LeCreateConnCancel>
            + ControllerCmdSync<LeReadBufferSize>
            + ControllerCmdSync<ReadBdAddr>
            + crate::SecurityCmds
            + crate::IsoStreamCmds
            + crate::SubratingCmds
            + crate::ShortConnIntervalCmds,
        C::Error: crate::fmt::Format,
    {
        let control_fut = self.control.run();
        let rx_fut = self.rx.run_with_handler(event_handler);
        let tx_fut = self.tx.run();
        pin_mut!(control_fut, rx_fut, tx_fut);
        match select3(&mut tx_fut, &mut rx_fut, &mut control_fut).await {
            Either3::First(result) => {
                trace!("[host] tx_fut exit");
                result
            }
            Either3::Second(result) => {
                trace!("[host] rx_fut exit");
                result
            }
            Either3::Third(result) => {
                trace!("[host] control_fut exit");
                result
            }
        }
    }
}

impl<'d, C: Controller, P: PacketPool> RxRunner<'d, C, P> {
    /// Run the receive loop that polls the controller for events.
    pub async fn run(&mut self) -> Result<(), BleHostError<C::Error>>
    where
        C: ControllerCmdSync<Disconnect>,
    {
        let dummy = DummyHandler;
        self.run_with_handler(&dummy).await
    }

    /// Runs the receive loop that pools the controller for events, dispatching
    /// vendor events to the provided closure.
    pub async fn run_with_handler<E: EventHandler>(&mut self, event_handler: &E) -> Result<(), BleHostError<C::Error>>
    where
        C: ControllerCmdSync<Disconnect>,
    {
        let host = &self.host;
        // use embassy_time::Instant;
        // let mut last = Instant::now();
        loop {
            // Task handling receiving data from the controller.
            let mut rx = host.controller.alloc_buf().map_err(BleHostError::Controller)?;
            // let now = Instant::now();
            // let elapsed = (now - last).as_millis();
            // if elapsed >= 1 {
            //     trace!("[host] time since last poll was {} us", elapsed);
            // }
            let result = host.controller.read(&mut rx).await;
            // last = Instant::now();
            //        trace!("[host] polling took {} ms", (polled - started).as_millis());
            match result {
                Ok(ControllerToHostPacket::Acl(acl)) => match host.handle_acl(acl, event_handler) {
                    Ok(_) => {}
                    Err(e) => {
                        warn!(
                            "[host] encountered error processing ACL data for {:?}: {:?}",
                            acl.handle(),
                            e
                        );

                        match e {
                            Error::InvalidState | Error::Disconnected => {
                                warn!("[host] requesting {:?} to be disconnected", acl.handle());
                                host.state.connections.log_status(true);
                                host.state.connections.request_handle_disconnect(
                                    acl.handle(),
                                    DisconnectReason::RemoteUserTerminatedConn,
                                );
                            }
                            _ => {}
                        }

                        let mut m = host.state.metrics.borrow_mut();
                        m.rx_errors = m.rx_errors.wrapping_add(1);
                    }
                },
                Ok(ControllerToHostPacket::Event(event)) => {
                    match event.kind {
                        EventKind::Le => {
                            let event = unwrap!(LeEventPacket::from_hci_bytes_complete(event.data));
                            match event.kind {
                                LeEventKind::LeConnectionComplete => {
                                    let e = unwrap!(LeConnectionComplete::from_hci_bytes_complete(event.data));
                                    if !host.handle_connection(
                                        e.status,
                                        e.handle,
                                        Address::new(e.peer_addr_kind, e.peer_addr),
                                        e.role,
                                        ConnParams {
                                            conn_interval: Duration::from_micros(e.conn_interval.as_micros()),
                                            peripheral_latency: e.peripheral_latency,
                                            supervision_timeout: Duration::from_micros(
                                                e.supervision_timeout.as_micros(),
                                            ),
                                        },
                                        #[cfg(feature = "security")]
                                        ResolvablePrivateAddrs::none(),
                                    ) {
                                        let _ = host
                                            .command(Disconnect::new(
                                                e.handle,
                                                DisconnectReason::RemoteDeviceTerminatedConnLowResources,
                                            ))
                                            .await;
                                        host.state.connect_command_state.canceled();
                                    }
                                }
                                LeEventKind::LeEnhancedConnectionComplete => {
                                    let e = unwrap!(LeEnhancedConnectionComplete::from_hci_bytes_complete(event.data));
                                    if !host.handle_connection(
                                        e.status,
                                        e.handle,
                                        Address::new(e.peer_addr_kind, e.peer_addr),
                                        e.role,
                                        ConnParams {
                                            conn_interval: Duration::from_micros(e.conn_interval.as_micros()),
                                            peripheral_latency: e.peripheral_latency,
                                            supervision_timeout: Duration::from_micros(
                                                e.supervision_timeout.as_micros(),
                                            ),
                                        },
                                        #[cfg(feature = "security")]
                                        ResolvablePrivateAddrs {
                                            local: Some(e.local_resolvable_private_addr).filter(|a| *a.raw() != [0; 6]),
                                            peer: Some(e.peer_resolvable_private_addr).filter(|a| *a.raw() != [0; 6]),
                                        },
                                    ) {
                                        let _ = host
                                            .command(Disconnect::new(
                                                e.handle,
                                                DisconnectReason::RemoteDeviceTerminatedConnLowResources,
                                            ))
                                            .await;
                                        host.state.connect_command_state.canceled();
                                    }
                                }
                                LeEventKind::LeScanTimeout => {
                                    #[cfg(feature = "scan")]
                                    host.state.scan_timeout.signal(());
                                }
                                LeEventKind::LeAdvertisingSetTerminated => {
                                    let set = unwrap!(LeAdvertisingSetTerminated::from_hci_bytes_complete(event.data));
                                    host.state.advertise_state.terminate(set.adv_handle);
                                }
                                LeEventKind::LeExtendedAdvertisingReport => {
                                    #[cfg(feature = "scan")]
                                    {
                                        let data =
                                            unwrap!(LeExtendedAdvertisingReport::from_hci_bytes_complete(event.data));
                                        event_handler.on_ext_adv_reports(data.reports.iter());
                                    }
                                }
                                LeEventKind::LeAdvertisingReport => {
                                    #[cfg(feature = "scan")]
                                    {
                                        let data = unwrap!(LeAdvertisingReport::from_hci_bytes_complete(event.data));
                                        event_handler.on_adv_reports(data.reports.iter());
                                    }
                                }
                                LeEventKind::LeLongTermKeyRequest => {
                                    host.state.connections.handle_security_hci_le_event(event)?;
                                }
                                LeEventKind::LePhyUpdateComplete => {
                                    let event = unwrap!(LePhyUpdateComplete::from_hci_bytes_complete(event.data));
                                    if let Err(e) = event.status.to_result() {
                                        warn!("[host] error updating phy for {:?}: {:?}", event.handle, e);
                                    } else {
                                        let _ = host.state.connections.post_handle_event(
                                            event.handle,
                                            ConnectionEvent::PhyUpdated {
                                                tx_phy: event.tx_phy,
                                                rx_phy: event.rx_phy,
                                            },
                                        );
                                    }
                                }
                                LeEventKind::LeConnectionUpdateComplete => {
                                    let event =
                                        unwrap!(LeConnectionUpdateComplete::from_hci_bytes_complete(event.data));
                                    if let Err(e) = event.status.to_result() {
                                        warn!(
                                            "[host] error updating connection parameters for {:?}: {:?}",
                                            event.handle, e
                                        );
                                    } else {
                                        let _ = host.state.connections.post_handle_event(
                                            event.handle,
                                            ConnectionEvent::ConnectionParamsUpdated {
                                                conn_interval: Duration::from_micros(event.conn_interval.as_micros()),
                                                peripheral_latency: event.peripheral_latency,
                                                supervision_timeout: Duration::from_micros(
                                                    event.supervision_timeout.as_micros(),
                                                ),
                                            },
                                        );
                                    }
                                }
                                LeEventKind::LeDataLengthChange => {
                                    let event = unwrap!(LeDataLengthChange::from_hci_bytes_complete(event.data));
                                    let _ = host.state.connections.post_handle_event(
                                        event.handle,
                                        ConnectionEvent::DataLengthUpdated {
                                            max_tx_octets: event.max_tx_octets,
                                            max_tx_time: event.max_tx_time,
                                            max_rx_octets: event.max_rx_octets,
                                            max_rx_time: event.max_rx_time,
                                        },
                                    );
                                }
                                LeEventKind::LeFrameSpaceUpdateComplete => {
                                    let event =
                                        unwrap!(LeFrameSpaceUpdateComplete::from_hci_bytes_complete(event.data));
                                    if let Err(e) = event.status.to_result() {
                                        warn!("[host] error updating frame space for {:?}: {:?}", event.handle, e);
                                    } else {
                                        let _ = host.state.connections.post_handle_event(
                                            event.handle,
                                            ConnectionEvent::FrameSpaceUpdated {
                                                frame_space: Duration::from_micros(event.frame_space.as_micros()),
                                                initiator: event.initiator,
                                                phys: event.phys,
                                                spacing_types: event.spacing_types,
                                            },
                                        );
                                    }
                                }
                                #[cfg(feature = "subrating")]
                                LeEventKind::LeSubrateChange => {
                                    let event = unwrap!(LeSubrateChange::from_hci_bytes_complete(event.data));
                                    if let Err(e) = event.status.to_result() {
                                        warn!("[host] error in subrate change for {:?}: {:?}", event.handle, e);
                                    } else {
                                        let _ = host.state.connections.post_handle_event(
                                            event.handle,
                                            ConnectionEvent::SubratingParamsUpdated {
                                                subrate_factor: event.subrate_factor,
                                                peripheral_latency: event.peripheral_latency,
                                                continuation_number: event.continuation_number,
                                                supervision_timeout: Duration::from_micros(
                                                    event.supervision_timeout.as_micros(),
                                                ),
                                            },
                                        );
                                    }
                                }
                                LeEventKind::LeConnectionRateChange => {
                                    let event = unwrap!(LeConnectionRateChange::from_hci_bytes_complete(event.data));
                                    if let Err(e) = event.status.to_result() {
                                        warn!("[host] error in connection rate change for {:?}: {:?}", event.handle, e);
                                    } else {
                                        let _ = host.state.connections.post_handle_event(
                                            event.handle,
                                            ConnectionEvent::ConnectionRateChanged {
                                                conn_interval: Duration::from_micros(event.conn_interval.as_micros()),
                                                subrate_factor: event.subrate_factor,
                                                peripheral_latency: event.peripheral_latency,
                                                continuation_number: event.continuation_number,
                                                supervision_timeout: Duration::from_micros(
                                                    event.supervision_timeout.as_micros(),
                                                ),
                                            },
                                        );
                                    }
                                }
                                LeEventKind::LeRemoteConnectionParameterRequest => {
                                    let event = unwrap!(LeRemoteConnectionParameterRequest::from_hci_bytes_complete(
                                        event.data
                                    ));
                                    let req = ConnectionParamsRequest::new(
                                        RequestedConnParams {
                                            min_connection_interval: Duration::from_micros(
                                                event.interval_min.as_micros(),
                                            ),
                                            max_connection_interval: Duration::from_micros(
                                                event.interval_max.as_micros(),
                                            ),
                                            max_latency: event.max_latency,
                                            supervision_timeout: Duration::from_micros(event.timeout.as_micros()),
                                            ..Default::default()
                                        },
                                        event.handle,
                                        #[cfg(feature = "connection-params-update")]
                                        false,
                                    );
                                    let _ = host
                                        .state
                                        .connections
                                        .post_handle_event(event.handle, ConnectionEvent::RequestConnectionParams(req));
                                }
                                #[cfg(feature = "iso")]
                                LeEventKind::LeCisRequest => {
                                    let e = unwrap!(LeCisRequest::from_hci_bytes_complete(event.data));
                                    event_handler.on_cis_request(&e);
                                }
                                #[cfg(feature = "iso")]
                                LeEventKind::LeCisEstablished => {
                                    let e = unwrap!(LeCisEstablished::from_hci_bytes_complete(event.data));
                                    event_handler.on_cis_established(&e);
                                }
                                _ => {
                                    warn!("Unknown LE event!");
                                }
                            }
                        }
                        EventKind::DisconnectionComplete => {
                            let e = unwrap!(DisconnectionComplete::from_hci_bytes_complete(event.data));
                            let handle = e.handle;
                            let reason = if let Err(e) = e.status.to_result() {
                                info!("[host] disconnection event on handle {}, status: {:?}", handle.raw(), e);
                                None
                            } else if let Err(err) = e.reason.to_result() {
                                info!(
                                    "[host] disconnection event on handle {}, reason: {:?}",
                                    handle.raw(),
                                    err
                                );
                                Some(e.reason)
                            } else {
                                info!("[host] disconnection event on handle {}", handle.raw());
                                None
                            }
                            .unwrap_or(Status::UNSPECIFIED);
                            let _ = host.state.connections.disconnected(handle, reason);
                            let _ = host.state.channels.disconnected(handle);
                            let mut m = host.state.metrics.borrow_mut();
                            m.disconnect_events = m.disconnect_events.wrapping_add(1);
                        }
                        EventKind::NumberOfCompletedPackets => {
                            let c = unwrap!(NumberOfCompletedPackets::from_hci_bytes_complete(event.data));
                            // Explicitly ignoring for now
                            for entry in c.completed_packets.iter() {
                                match (entry.handle(), entry.num_completed_packets()) {
                                    (Ok(handle), Ok(completed)) => {
                                        let _ = host.state.connections.confirm_sent(handle, completed as usize);
                                        event_handler.on_packets_completed(completed as usize);
                                    }
                                    (Ok(handle), Err(e)) => {
                                        warn!("[host] error processing completed packets for {:?}: {:?}", handle, e);
                                    }
                                    _ => {}
                                }
                            }
                        }
                        EventKind::Vendor => {
                            let vendor = unwrap!(Vendor::from_hci_bytes_complete(event.data));
                            event_handler.on_vendor(&vendor);
                        }
                        EventKind::EncryptionChangeV1 | EventKind::EncryptionKeyRefreshComplete => {
                            host.state.connections.handle_security_hci_event(event)?;
                        }
                        // Ignore
                        _ => {}
                    }
                }
                #[cfg(feature = "iso")]
                Ok(ControllerToHostPacket::Iso(packet)) => {
                    event_handler.on_iso_data(&packet);
                }
                // Ignore
                Ok(_) => {}
                Err(e) => {
                    return Err(BleHostError::Controller(e));
                }
            }
        }
    }
}

enum CancelledCommandState {
    #[cfg(feature = "central")]
    Connect(bool),
    #[cfg(feature = "peripheral")]
    Advertise(bool),
    #[cfg(feature = "scan")]
    Scan(bool),
    #[cfg(feature = "security")]
    SyncResolvingList(ResolvingListUpdate),
    #[cfg(all(feature = "security", feature = "central"))]
    RotateRpa,
}

impl<'d, C: Controller, P: PacketPool> ControlRunner<'d, C, P> {
    /// Run the control loop for the host
    pub async fn run(&mut self) -> Result<(), BleHostError<C::Error>>
    where
        C: ControllerCmdSync<Disconnect>
            + ControllerCmdSync<SetEventMask>
            + ControllerCmdSync<SetEventMaskPage2>
            + ControllerCmdSync<LeSetEventMask>
            + ControllerCmdSync<LeSetRandomAddr>
            + ControllerCmdSync<HostBufferSize>
            + ControllerCmdAsync<LeConnUpdate>
            + ControllerCmdSync<LeReadFilterAcceptListSize>
            + ControllerCmdSync<SetControllerToHostFlowControl>
            + ControllerCmdSync<Reset>
            + ControllerCmdSync<LeCreateConnCancel>
            + for<'t> ControllerCmdSync<LeSetAdvEnable>
            + for<'t> ControllerCmdSync<LeSetExtAdvEnable<'t>>
            + ControllerCmdSync<LeSetScanEnable>
            + ControllerCmdSync<LeSetExtScanEnable>
            + for<'t> ControllerCmdSync<HostNumberOfCompletedPackets<'t>>
            + ControllerCmdSync<LeReadBufferSize>
            + ControllerCmdSync<ReadBdAddr>
            + crate::SecurityCmds
            + crate::IsoStreamCmds
            + crate::SubratingCmds
            + crate::ShortConnIntervalCmds,
        C::Error: crate::fmt::Format,
    {
        let host = &self.host;
        Reset::new().exec(host.controller).await?;

        #[cfg(feature = "security")]
        {
            let mut seed = [0u8; 32];
            for chunk in seed.chunks_mut(8) {
                let bytes: [u8; 8] = LeRand::new().exec(host.controller).await?;
                chunk.copy_from_slice(&bytes);
            }
            host.state.connections.security_manager.set_random_generator_seed(seed);
        }

        {
            let addr = host.state.address.map(|a| a.addr);

            #[cfg(all(feature = "security", feature = "central"))]
            let addr = host.state.connections.security_manager.generate_local_rpa().or(addr);

            if let Some(addr) = addr {
                LeSetRandomAddr::new(addr).exec(host.controller).await?;

                #[cfg(all(feature = "security", feature = "central"))]
                if host.is_privacy_enabled() {
                    host.state
                        .rpa_expires_at
                        .set(Instant::now() + host.state.rpa_timeout.get());
                }
            }
        }

        SetEventMask::new(
            EventMask::new()
                .enable_le_meta(true)
                .enable_conn_request(true)
                .enable_conn_complete(true)
                .enable_hardware_error(true)
                .enable_disconnection_complete(true)
                .enable_encryption_change_v1(true)
                .enable_encryption_key_refresh_complete(true),
        )
        .exec(host.controller)
        .await?;

        if let Err(e) = SetEventMaskPage2::new(EventMaskPage2::new().enable_encryption_change_v2(true))
            .exec(host.controller)
            .await
        {
            match e {
                cmd::Error::Hci(bt_hci::param::Error::UNKNOWN_CMD) => warn!("set event mask page 2 is not supported"),
                e => Err(e)?,
            }
        }

        let mask = LeEventMask::new()
            .enable_le_conn_complete(true)
            .enable_le_enhanced_conn_complete_v1(true)
            .enable_le_conn_update_complete(true)
            .enable_le_adv_set_terminated(true)
            .enable_le_adv_report(true)
            .enable_le_scan_timeout(true)
            .enable_le_ext_adv_report(true)
            .enable_le_long_term_key_request(true)
            .enable_le_phy_update_complete(true)
            .enable_le_data_length_change(true);

        #[cfg(feature = "subrating")]
        let mask = mask.enable_le_subrate_change(true);

        #[cfg(feature = "iso")]
        let mask = mask.enable_le_cis_established_v1(true).enable_le_cis_request(true);

        #[cfg(feature = "connection-params-update")]
        let mask = mask.enable_le_remote_conn_parameter_request(true);

        #[cfg(feature = "shorter-connection-intervals")]
        let mask = mask.enable_le_connection_rate_change(true);

        LeSetEventMask::new(mask).exec(host.controller).await?;

        // Without the Connection Isochronous Stream (Host Support) bit set, a peer central is not allowed to
        // create or accept a CIS.
        #[cfg(feature = "iso")]
        {
            const LE_FEATURE_CIS_HOST: u8 = 32;
            if let Err(e) = LeSetHostFeature::new(LE_FEATURE_CIS_HOST, 1)
                .exec(host.controller)
                .await
            {
                match e {
                    cmd::Error::Hci(bt_hci::param::Error::UNSUPPORTED | bt_hci::param::Error::UNKNOWN_CMD) => {
                        warn!("[host] connection isochronous streams are not supported")
                    }
                    e => Err(e)?,
                }
            }
        }

        // Without the Connection Subrating (Host Support) bit set, a peer central is not allowed to
        // start the Connection Subrate Update procedure on us.
        #[cfg(feature = "subrating")]
        {
            const LE_FEATURE_CONN_SUBRATING_HOST: u8 = 38;
            if let Err(e) = LeSetHostFeature::new(LE_FEATURE_CONN_SUBRATING_HOST, 1)
                .exec(host.controller)
                .await
            {
                match e {
                    cmd::Error::Hci(bt_hci::param::Error::UNSUPPORTED | bt_hci::param::Error::UNKNOWN_CMD) => {
                        warn!("[host] connection subrating is not supported")
                    }
                    e => Err(e)?,
                }
            }
        }

        // Without the Shorter Connection Intervals (Host Support) bit set, a peer central is not allowed to
        // start the Connection Rate Change procedure on us.
        #[cfg(feature = "shorter-connection-intervals")]
        {
            const LE_FEATURE_SHORTER_CONNECTION_INTERVALS_HOST: u16 = 73;
            if let Err(e) = LeSetHostFeatureV2::new(LE_FEATURE_SHORTER_CONNECTION_INTERVALS_HOST, 1)
                .exec(host.controller)
                .await
            {
                match e {
                    cmd::Error::Hci(bt_hci::param::Error::UNSUPPORTED | bt_hci::param::Error::UNKNOWN_CMD) => {
                        warn!("[host] shorter connection intervals are not supported")
                    }
                    e => Err(e)?,
                }
            }
        }

        info!(
            "[host] using packet pool with MTU {} capacity {}",
            P::MTU,
            P::capacity(),
        );

        let ret = LeReadFilterAcceptListSize::new().exec(host.controller).await?;
        info!("[host] filter accept list size: {}", ret);

        let ret = LeReadBufferSize::new().exec(host.controller).await?;
        info!(
            "[host] setting txq to {}, fragmenting at {}",
            ret.total_num_le_acl_data_packets as usize, ret.le_acl_data_packet_length as usize
        );
        host.state
            .connections
            .set_link_credits(ret.total_num_le_acl_data_packets as usize);

        {
            const ACL_LEN: u16 = 255;
            const ACL_N: u16 = 1;
            info!(
                "[host] configuring host buffers ({} packets of size {})",
                ACL_N, ACL_LEN,
            );
            if let Err(_e) = HostBufferSize::new(ACL_LEN, 0, ACL_N, 0).exec(host.controller).await {
                warn!("[host] error configuring host buffers (continuing)");
            }
        }

        /*
                #[cfg(feature = "controller-host-flow-control")]
                {
                    info!("[host] enabling flow control");
                    SetControllerToHostFlowControl::new(ControllerToHostFlowControl::AclOnSyncOff)
                        .exec(&host.controller)
                        .await?;
                }
        */

        let _ = host.state.initialized.init(InitialState {
            acl_max: ret.le_acl_data_packet_length as usize,
            acl_total: ret.total_num_le_acl_data_packets as usize,
        });
        info!("[host] initialized");

        let device_address = host.command(ReadBdAddr::new()).await?;
        if *device_address.raw() != [0, 0, 0, 0, 0, 0] {
            let device_address = Address::new(AddrKind::PUBLIC, device_address);
            info!("[host] Device Address {}", device_address);
            if host.state.address.is_none() {
                #[cfg(feature = "security")]
                host.state
                    .connections
                    .security_manager
                    .set_local_address(device_address);
            }
        }

        // Set default RPA timeout in controller
        #[cfg(feature = "security")]
        {
            let timeout_secs = host.state.rpa_timeout.get().as_secs();
            LeSetResolvablePrivateAddrTimeout::new(bt_hci::param::Duration::from_secs(timeout_secs as u32))
                .exec(host.controller)
                .await?;
            info!("[host] RPA timeout set to {}s", timeout_secs);
        }

        // Initialize privacy: sync resolving list
        #[cfg(feature = "security")]
        if host.is_privacy_enabled() {
            host.state.resolving_list_state.borrow_mut().clear();
            host.sync_resolving_list(ResolvingListUpdate::FullSync).await?;
            info!("[host] privacy initialized");
        }

        loop {
            match select5(
                poll_fn(|cx| host.state.connections.poll_disconnecting(Some(cx))),
                poll_fn(|cx| host.state.channels.poll_disconnecting(Some(cx))),
                poll_fn(|cx| host.poll_cancelled(cx)),
                #[cfg(feature = "security")]
                {
                    host.state.connections.poll_security_events()
                },
                #[cfg(not(feature = "security"))]
                {
                    core::future::pending::<()>()
                },
                #[cfg(all(feature = "security", feature = "central"))]
                {
                    host.wait_for_rpa_expiration()
                },
                #[cfg(not(all(feature = "security", feature = "central")))]
                {
                    core::future::pending::<()>()
                },
            )
            .await
            {
                Either5::First(request) => {
                    trace!("[host] poll disconnecting links");
                    match host.command(Disconnect::new(request.handle(), request.reason())).await {
                        Ok(_) => {}
                        Err(BleHostError::BleHost(Error::Hci(bt_hci::param::Error::UNKNOWN_CONN_IDENTIFIER))) => {}
                        Err(BleHostError::BleHost(Error::NotFound)) => {}
                        Err(BleHostError::BleHost(Error::Disconnected)) => {}
                        Err(e) => {
                            return Err(e);
                        }
                    }
                    request.confirm();
                }
                Either5::Second(request) => {
                    trace!("[host] poll disconnecting channels");
                    match request.send(host).await {
                        Ok(_) => {}
                        Err(BleHostError::BleHost(Error::Hci(bt_hci::param::Error::UNKNOWN_CONN_IDENTIFIER))) => {}
                        Err(BleHostError::BleHost(Error::NotFound)) => {}
                        Err(BleHostError::BleHost(Error::Disconnected)) => {}
                        Err(e) => {
                            return Err(e);
                        }
                    }
                    request.confirm();
                }
                Either5::Third(action) => match action {
                    #[cfg(feature = "central")]
                    CancelledCommandState::Connect(_) => {
                        trace!("[host] cancel connection create");
                        if let Err(err) = host.command(LeCreateConnCancel::new()).await {
                            warn!("[host] error cancelling connection: {:?}", err);
                        }
                        // Signal to ensure no one is stuck
                        host.state.connect_command_state.canceled();
                    }
                    #[cfg(feature = "peripheral")]
                    CancelledCommandState::Advertise(ext) => {
                        trace!("[host] disabling advertising");
                        #[cfg(feature = "extended-advertising")]
                        if ext {
                            host.command(LeSetExtAdvEnable::new(false, &[])).await?
                        } else {
                            host.command(LeSetAdvEnable::new(false)).await?
                        }
                        #[cfg(not(feature = "extended-advertising"))]
                        {
                            // Extended advertising cannot have been started
                            // (`advertise_ext` is compiled out), so `ext` is
                            // always false here.
                            let _ = ext;
                            host.command(LeSetAdvEnable::new(false)).await?
                        }
                        host.state.advertise_command_state.canceled();
                    }
                    #[cfg(feature = "scan")]
                    CancelledCommandState::Scan(ext) => {
                        trace!("[host] disabling scanning");
                        if ext {
                            // TODO: A bit opinionated but not more than before
                            host.command(LeSetExtScanEnable::new(
                                false,
                                FilterDuplicates::Disabled,
                                bt_hci::param::Duration::from_secs(0),
                                bt_hci::param::Duration::from_secs(0),
                            ))
                            .await?;
                        } else {
                            host.command(LeSetScanEnable::new(false, false)).await?;
                        }
                        host.state.scan_command_state.canceled();
                    }
                    #[cfg(feature = "security")]
                    CancelledCommandState::SyncResolvingList(update) => {
                        host.sync_resolving_list(update).await?;
                    }
                    #[cfg(all(feature = "security", feature = "central"))]
                    CancelledCommandState::RotateRpa => {
                        host.rotate_rpa().await?;
                    }
                },
                Either5::Fourth(request) => {
                    #[cfg(feature = "security")]
                    {
                        let event_data = request.unwrap_or(SecurityEventData::Timeout);
                        host.state.connections.handle_security_event(host, event_data).await?;
                    }
                }
                Either5::Fifth(()) => {
                    #[cfg(all(feature = "security", feature = "central"))]
                    {
                        host.rotate_rpa().await?;
                    }
                }
            }
        }
    }
}

impl<'d, C: Controller, P: PacketPool> TxRunner<'d, C, P> {
    /// Run the transmit loop for the host.
    pub async fn run(&mut self) -> Result<(), BleHostError<C::Error>> {
        let host = &self.host;
        let params = host.state.initialized.get().await;
        loop {
            let (conn, pdu) = host.state.connections.outbound().await;
            match host.l2cap_pdu(conn).await {
                Ok(mut sender) => match sender.send(pdu.as_ref()).await {
                    Ok(()) => {}
                    Err(BleHostError::BleHost(Error::NotFound)) | Err(BleHostError::BleHost(Error::Disconnected)) => {
                        warn!("[host] link disconnected while sending outbound pdu (ignored)");
                    }
                    Err(e) => {
                        warn!("[host] error sending outbound pdu");
                        return Err(e);
                    }
                },
                Err(BleHostError::BleHost(Error::NotFound)) => {
                    warn!("[host] unable to send data to disconnected host (ignored)");
                }
                Err(BleHostError::BleHost(Error::Disconnected)) => {
                    warn!("[host] unable to send data to disconnected host (ignored)");
                }
                Err(e) => {
                    warn!("[host] error requesting sending outbound pdu");
                    return Err(e);
                }
            }
        }
    }
}

pub struct L2capSender<'a, T: Controller, P: PacketPool> {
    pub(crate) controller: &'a T,
    pub(crate) handle: ConnHandle,
    pub(crate) acl_send_lock: AclSendLock<'a, P>,
    pub(crate) fragment_size: u16,
    pub(crate) max_fragments: usize,
}

impl<'a, T: Controller, P: PacketPool> L2capSender<'a, T, P> {
    pub(crate) fn try_send(&mut self, pdu: &[u8]) -> Result<(), BleHostError<T::Error>>
    where
        T: blocking::Controller,
    {
        let n_acl = pdu.chunks(self.fragment_size as usize).len();
        if n_acl > self.max_fragments {
            return Err(Error::NoPermits.into());
        }

        let mut grant = match self.acl_send_lock.poll_request_to_send(n_acl, None) {
            Poll::Ready(res) => res?,
            Poll::Pending => return Err(Error::Busy.into()),
        };
        let mut pbf = AclPacketBoundary::FirstNonFlushable;
        //info!(
        //    "[host] fragmenting PDU of size {} into {} sized fragments",
        //    pdu.len(),
        //    self.fragment_size
        //);
        for chunk in pdu.chunks(self.fragment_size as usize) {
            let acl = AclPacket::new(self.handle, pbf, AclBroadcastFlag::PointToPoint, chunk);
            match self.controller.try_write_acl_data(&acl) {
                Ok(_result) => {
                    grant.confirm(1);
                    trace!("[host] sent acl packet len = {}", chunk.len());
                }
                Err(blocking::TryError::Busy) => {
                    warn!("hci: acl data send busy");
                    return Err(Error::Busy.into());
                }
                Err(blocking::TryError::Error(e)) => return Err(BleHostError::Controller(e)),
            }
            pbf = AclPacketBoundary::Continuing;
        }
        Ok(())
    }

    pub(crate) async fn send(&mut self, pdu: &[u8]) -> Result<(), BleHostError<T::Error>> {
        //info!(
        //    "[host] fragmenting PDU of size {} into {} sized fragments",
        //    pdu.len(),
        //    self.fragment_size
        //);
        let mut pbf = AclPacketBoundary::FirstNonFlushable;
        for chunk in pdu.chunks(self.fragment_size as usize) {
            let mut grant = poll_fn(|cx| self.acl_send_lock.poll_request_to_send(1, Some(cx))).await?;
            let acl = AclPacket::new(self.handle, pbf, AclBroadcastFlag::PointToPoint, chunk);
            self.controller
                .write_acl_data(&acl)
                .await
                .map_err(BleHostError::Controller)?;
            grant.confirm(1);
            pbf = AclPacketBoundary::Continuing;
            trace!("[host] sent acl packet len = {}", chunk.len());
        }
        Ok(())
    }
}

/// A type to delay the drop handler invocation.
#[must_use = "to delay the drop handler invocation to the end of the scope"]
pub struct OnDrop<F: FnOnce()> {
    f: MaybeUninit<F>,
}

impl<F: FnOnce()> OnDrop<F> {
    /// Create a new instance.
    pub fn new(f: F) -> Self {
        Self { f: MaybeUninit::new(f) }
    }

    /// Prevent drop handler from running.
    pub fn defuse(self) {
        core::mem::forget(self)
    }
}

impl<F: FnOnce()> Drop for OnDrop<F> {
    fn drop(&mut self) {
        unsafe { self.f.as_ptr().read()() }
    }
}