feat(gatt): support multiple gatt client

Signed-off-by: Haobo Gu <[email protected]>

Author: HaoboGu

host/src/connection.rs

@@ -271,6 +271,11 @@ impl<'stack, P: PacketPool> Connection<'stack, P> {
         self.manager.next_gatt(self.index).await
     }
 
+    #[cfg(feature = "gatt")]
+    pub(crate) async fn next_gatt_client(&self) -> Pdu<P::Packet> {
+        self.manager.next_gatt_client(self.index).await
+    }
+
     /// Check if still connected
     pub fn is_connected(&self) -> bool {
         self.manager.is_connected(self.index)

host/src/connection_manager.rs

@@ -133,6 +133,24 @@ impl<'d, P: PacketPool> ConnectionManager<'d, P> {
         })
     }
 
+    #[cfg(feature = "gatt")]
+    pub(crate) fn post_gatt_client(&self, handle: ConnHandle, pdu: Pdu<P::Packet>) -> Result<(), Error> {
+        self.with_mut(|state| {
+            for entry in state.connections.iter() {
+                if entry.state == ConnectionState::Connected && Some(handle) == entry.handle {
+                    entry.gatt_client.try_send(pdu).map_err(|_| Error::OutOfMemory)?;
+                    return Ok(());
+                }
+            }
+            Err(Error::NotFound)
+        })
+    }
+
+    #[cfg(feature = "gatt")]
+    pub(crate) async fn next_gatt_client(&self, index: u8) -> Pdu<P::Packet> {
+        poll_fn(|cx| self.with_mut(|state| state.connections[index as usize].gatt_client.poll_receive(cx))).await
+    }
+
     pub(crate) fn peer_address(&self, index: u8) -> BdAddr {
         self.with_mut(|state| {
             let state = &mut state.connections[index as usize];
@@ -791,6 +809,8 @@ pub struct ConnectionStorage<P> {
     pub reassembly: PacketReassembly<P>,
     #[cfg(feature = "gatt")]
     pub gatt: GattChannel<P>,
+    #[cfg(feature = "gatt")]
+    pub(crate) gatt_client: GattChannel<P>,
 }
 
 /// Connection metrics
@@ -874,6 +894,8 @@ impl<P> ConnectionStorage<P> {
             events: EventChannel::new(),
             #[cfg(feature = "gatt")]
             gatt: GattChannel::new(),
+            #[cfg(feature = "gatt")]
+            gatt_client: GattChannel::new(),
             reassembly: PacketReassembly::new(),
             #[cfg(feature = "security")]
             bondable: false,

host/src/gatt.rs

@@ -9,7 +9,7 @@ use bt_hci::uuid::declarations::{CHARACTERISTIC, PRIMARY_SERVICE};
 use bt_hci::uuid::descriptors::CLIENT_CHARACTERISTIC_CONFIGURATION;
 use embassy_futures::select::{select, Either};
 use embassy_sync::blocking_mutex::raw::{NoopRawMutex, RawMutex};
-use embassy_sync::channel::{Channel, DynamicReceiver};
+use embassy_sync::channel::Channel;
 use embassy_sync::pubsub::{self, PubSubChannel, WaitResult};
 use embassy_time::Duration;
 use heapless::Vec;
@@ -650,7 +650,6 @@ const NOTIF_QSIZE: usize = config::GATT_CLIENT_NOTIFICATION_QUEUE_SIZE;
 /// A GATT client capable of using the GATT protocol.
 pub struct GattClient<'reference, T: Controller, P: PacketPool, const MAX_SERVICES: usize> {
     known_services: RefCell<Vec<ServiceHandle, MAX_SERVICES>>,
-    rx: DynamicReceiver<'reference, (ConnHandle, Pdu<P::Packet>)>,
     stack: &'reference Stack<'reference, T, P>,
     connection: Connection<'reference, P>,
     response_channel: Channel<NoopRawMutex, (ConnHandle, Pdu<P::Packet>), 1>,
@@ -754,7 +753,6 @@ impl<'reference, C: Controller, P: PacketPool, const MAX_SERVICES: usize> GattCl
         connection.send(Pdu::new(buf, len)).await;
         Ok(Self {
             known_services: RefCell::new(heapless::Vec::new()),
-            rx: stack.host.att_client.receiver().into(),
             stack,
             connection: connection.clone(),
 
@@ -1173,7 +1171,8 @@ impl<'reference, C: Controller, P: PacketPool, const MAX_SERVICES: usize> GattCl
     /// Task which handles GATT rx data (needed for notifications to work)
     pub async fn task(&self) -> Result<(), BleHostError<C::Error>> {
         loop {
-            let (handle, pdu) = self.rx.receive().await;
+            let handle = self.connection.handle();
+            let pdu = self.connection.next_gatt_client().await;
             let data = pdu.as_ref();
             // handle notifications
             if pdu.as_ref()[0] == ATT_HANDLE_VALUE_NTF {

host/src/host.rs

@@ -37,8 +37,6 @@ use bt_hci::{ControllerToHostPacket, FromHciBytes, WriteHci};
 use embassy_futures::select::{select3, select4, Either3, Either4};
 use embassy_sync::once_lock::OnceLock;
 use embassy_sync::waitqueue::WakerRegistration;
-#[cfg(feature = "gatt")]
-use embassy_sync::{blocking_mutex::raw::NoopRawMutex, channel::Channel};
 use embassy_time::Duration;
 use futures::pin_mut;
 
@@ -71,8 +69,6 @@ pub(crate) struct BleHost<'d, T, P: PacketPool> {
     pub(crate) controller: T,
     pub(crate) connections: ConnectionManager<'d, P>,
     pub(crate) channels: ChannelManager<'d, P>,
-    #[cfg(feature = "gatt")]
-    pub(crate) att_client: Channel<NoopRawMutex, (ConnHandle, Pdu<P::Packet>), { crate::config::L2CAP_RX_QUEUE_SIZE }>,
     pub(crate) advertise_state: AdvState<'d>,
     pub(crate) advertise_command_state: CommandState<bool>,
     pub(crate) connect_command_state: CommandState<bool>,
@@ -211,8 +207,6 @@ where
             controller,
             connections: ConnectionManager::new(connections, P::MTU as u16 - 4),
             channels: ChannelManager::new(channels),
-            #[cfg(feature = "gatt")]
-            att_client: Channel::new(),
             advertise_state: AdvState::new(advertise_handles),
             advertise_command_state: CommandState::new(),
             scan_command_state: CommandState::new(),
@@ -470,7 +464,7 @@ where
                             self.connections.post_gatt(acl.handle(), pdu)?;
                         }
                         Ok(att::Att::Server(_)) => {
-                            if let Err(e) = self.att_client.try_send((acl.handle(), pdu)) {
+                            if let Err(e) = self.connections.post_gatt_client(acl.handle(), pdu) {
                                 return Err(Error::OutOfMemory);
                             }
                         }