Merge "floss: Guarantee the execution order of GATT client callbacks" into main

use num_traits::cast::{FromPrimitive, ToPrimitive};

use rand::rngs::SmallRng;

use rand::{RngCore, SeedableRng};

use std::collections::{HashMap, HashSet};

use std::collections::{HashMap, HashSet, VecDeque};

use std::convert::{TryFrom, TryInto};

use std::sync::{Arc, Mutex, MutexGuard};

use tokio::sync::mpsc::Sender;

    pub fn init_profiles(&mut self, tx: Sender<Message>, api_tx: Sender<APIMessage>) {

        self.gatt = Gatt::new(&self.intf.lock().unwrap()).map(|gatt| Arc::new(Mutex::new(gatt)));

        // TODO(b/353643607): Make this dispatch_queue design more general for all profiles.

        let tx_clone = tx.clone();

        let async_mutex = Arc::new(tokio::sync::Mutex::new(()));

        let dispatch_queue = Arc::new(Mutex::new(VecDeque::new()));

        let gatt_client_callbacks_dispatcher = GattClientCallbacksDispatcher {

            dispatch: Box::new(move |cb| {

                let tx_clone = tx_clone.clone();

                let async_mutex = async_mutex.clone();

                let dispatch_queue = dispatch_queue.clone();

                // Enqueue the callbacks at the synchronized block to ensure the order.

                dispatch_queue.lock().unwrap().push_back(cb);

                topstack::get_runtime().spawn(async move {

                    // Acquire the lock first to ensure |pop_front| and |tx_clone.send| not

                    // interrupted by the other async threads.

                    let _guard = async_mutex.lock().await;

                    // Consume exactly one callback.

                    let cb = dispatch_queue.lock().unwrap().pop_front().unwrap();

                    let _ = tx_clone.send(Message::GattClient(cb)).await;

                });

            }),