// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use anyhow::{anyhow, format_err, Context as _, Error};
use async_helpers::hanging_get::asynchronous as hanging_get;
use fidl::endpoints::{Proxy, ServerEnd};
use fidl_fuchsia_bluetooth::{Appearance, DeviceClass};
use fidl_fuchsia_bluetooth_bredr::ProfileMarker;
use fidl_fuchsia_bluetooth_gatt::Server_Marker;
use fidl_fuchsia_bluetooth_gatt2::{
    LocalServiceRequest, Server_Marker as Server_Marker2, Server_Proxy,
};
use fidl_fuchsia_bluetooth_host::{DiscoverySessionProxy, HostProxy, ProtocolRequest};
use fidl_fuchsia_bluetooth_le::{CentralMarker, PeripheralMarker};
use fidl_fuchsia_bluetooth_sys::{
    self as sys, InputCapability, OutputCapability, PairingDelegateProxy,
};
use fuchsia_async::{self as fasync, DurationExt, TimeoutExt};
use fuchsia_bluetooth::inspect::{DebugExt, Inspectable, ToProperty};
use fuchsia_bluetooth::types::pairing_options::PairingOptions;
use fuchsia_bluetooth::types::{
    Address, BondingData, HostData, HostId, HostInfo, Identity, Peer, PeerId,
};
use fuchsia_inspect::{self as inspect, unique_name, NumericProperty, Property};
use fuchsia_inspect_contrib::{inspect_log, nodes::BoundedListNode};
use fuchsia_sync::RwLock;
use fuchsia_zircon::{self as zx, AsHandleRef, Duration};
use futures::channel::{mpsc, oneshot};
use futures::future::{self, BoxFuture, FusedFuture, Future, Shared};
use futures::FutureExt;
use slab::Slab;
use std::collections::HashMap;
use std::sync::{Arc, Weak};
use std::task::{Context, Poll, Waker};
use tracing::{debug, error, info, trace, warn};

use crate::{
    build_config, generic_access_service,
    host_device::{HostDevice, HostDiscoverableSession, HostListener},
    services::pairing::pairing_dispatcher::{PairingDispatcher, PairingDispatcherHandle},
    store::stash::Stash,
    types,
    watch_peers::PeerWatcher,
};

pub use fidl_fuchsia_device::DEFAULT_DEVICE_NAME;

/// Policies for HostDispatcher::set_name
#[derive(Copy, Clone, PartialEq, Debug)]
pub enum NameReplace {
    /// Keep the current name if it is already set, but set a new name if it hasn't been.
    Keep,
    /// Replace the current name unconditionally.
    Replace,
}

pub static HOST_INIT_TIMEOUT: i64 = 5; // Seconds

/// Available FIDL services that can be provided by a particular Host
#[derive(Copy, Clone)]
pub enum HostService {
    LeCentral,
    LePeripheral,
    LeGatt,
    LeGatt2,
    Profile,
}

/// When a client requests Discovery, we establish and store two distinct sessions; the dispatcher
/// DiscoverySession, an Arc<> of which is returned to clients and represents the dispatcher's
/// state of discovery that perists as long as one client maintains an Arc<> to the session, and
/// the DiscoverySessionProxy, which is returned by the active host device on which discovery is
/// physically occurring and persists until the host disappears or discovery is stopped.
pub enum DiscoveryState {
    NotDiscovering,
    Pending {
        // Additional client requests for discovery made while discovery is asynchronously
        // starting.
        session_receiver: Shared<oneshot::Receiver<Arc<DiscoverySession>>>,
        session_sender: oneshot::Sender<Arc<DiscoverySession>>,
        discovery_stopped_receiver: Shared<oneshot::Receiver<()>>,
        discovery_stopped_sender: oneshot::Sender<()>,
        start_discovery_task: fasync::Task<()>,
    },
    Discovering {
        session: Weak<DiscoverySession>,
        discovery_proxy: DiscoverySessionProxy,
        started: fasync::Time,
        discovery_on_closed_task: fasync::Task<()>,
        discovery_stopped_receiver: Shared<oneshot::Receiver<()>>,
        discovery_stopped_sender: oneshot::Sender<()>,
    },
    Stopping {
        // DiscoverySessionProxy needs to be held while stopping so that peer closed signal can be received.
        discovery_proxy: DiscoverySessionProxy,
        // Contains requests for discovery made while discovery is stopping (edge case).
        // This field is optional so we know if we don't need to restart discovery.
        session_receiver: Option<Shared<oneshot::Receiver<Arc<DiscoverySession>>>>,
        session_sender: Option<oneshot::Sender<Arc<DiscoverySession>>>,
        discovery_on_closed_task: fasync::Task<()>,
        discovery_stopped_receiver: Shared<oneshot::Receiver<()>>,
        discovery_stopped_sender: oneshot::Sender<()>,
    },
}

impl DiscoveryState {
    // Idempotently end the discovery session.
    // Returns the duration of a session, if one was ended.
    fn stop_discovery_session(&mut self) -> Option<fasync::Duration> {
        if let DiscoveryState::Discovering {
            discovery_proxy,
            started,
            discovery_on_closed_task,
            discovery_stopped_receiver,
            discovery_stopped_sender,
            ..
        } = std::mem::replace(self, DiscoveryState::NotDiscovering)
        {
            // stop() is synchronous, but stopping is an asynchronous procedure and will complete
            // when the Discovery event stream terminates.
            let _ = discovery_proxy.stop();
            *self = DiscoveryState::Stopping {
                discovery_proxy,
                session_receiver: None,
                session_sender: None,
                discovery_on_closed_task,
                discovery_stopped_receiver,
                discovery_stopped_sender,
            };
            return Some(fasync::Time::now() - started);
        }
        None
    }

    pub fn on_stopped(&mut self) -> impl FusedFuture<Output = ()> {
        let rx = match self {
            DiscoveryState::NotDiscovering => {
                let (tx, rx) = oneshot::channel();
                let _ = tx.send(());
                rx.shared()
            }
            DiscoveryState::Pending { discovery_stopped_receiver, .. } => {
                discovery_stopped_receiver.clone()
            }
            DiscoveryState::Discovering { discovery_stopped_receiver, .. } => {
                discovery_stopped_receiver.clone()
            }
            DiscoveryState::Stopping { discovery_stopped_receiver, .. } => {
                discovery_stopped_receiver.clone()
            }
        };
        rx.map(|_| ())
    }

    fn get_active_session_future(
        &mut self,
    ) -> Option<BoxFuture<'static, Result<Arc<DiscoverySession>, oneshot::Canceled>>> {
        match self {
            DiscoveryState::NotDiscovering => None,
            DiscoveryState::Pending { session_receiver, .. } => {
                Some(session_receiver.clone().boxed())
            }
            DiscoveryState::Discovering { session, .. } => {
                let session = session.upgrade().expect("session must exist in Discovering state");
                Some(future::ready(Ok(session)).boxed())
            }
            DiscoveryState::Stopping { session_receiver, session_sender, .. } => {
                match session_receiver {
                    Some(recv) => Some(recv.clone().boxed()),
                    None => {
                        let (send, recv) = oneshot::channel();
                        let recv = recv.shared();
                        *session_receiver = Some(recv.clone());
                        *session_sender = Some(send);
                        Some(recv.boxed())
                    }
                }
            }
        }
    }
}

impl std::fmt::Debug for DiscoveryState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DiscoveryState::NotDiscovering => {
                write!(f, "DiscoveryState::NotDiscovering")
            }
            DiscoveryState::Pending { .. } => {
                write!(f, "DiscoveryState::Pending")
            }
            DiscoveryState::Discovering { .. } => {
                write!(f, "DiscoveryState::Discovering")
            }
            DiscoveryState::Stopping { .. } => {
                write!(f, "DiscoveryState::Stopping")
            }
        }
    }
}

/// A dispatcher discovery session, which persists as long as at least one client holds an
/// Arc<> to it.
pub struct DiscoverySession {
    dispatcher_state: Arc<RwLock<HostDispatcherState>>,
}

impl DiscoverySession {
    pub fn on_discovery_end(&self) -> impl FusedFuture<Output = ()> {
        self.dispatcher_state.write().discovery.on_stopped()
    }
}

impl Drop for DiscoverySession {
    fn drop(&mut self) {
        let mut write = self.dispatcher_state.write();
        if let Some(dur) = write.discovery.stop_discovery_session() {
            inspect_log!(write.inspect.discovery_history, duration: dur.into_seconds_f64());
        }
    }
}

static RECENTLY_REMOVED_PEERS_COUNT: usize = 15;
static RECENT_DISCOVERY_SESSIONS_COUNT: usize = 5;

struct HostDispatcherInspect {
    _inspect: inspect::Node,
    peers: inspect::Node,
    hosts: inspect::Node,
    host_count: inspect::UintProperty,
    device_class: inspect::StringProperty,
    peer_count: inspect::UintProperty,
    input_capability: inspect::StringProperty,
    output_capability: inspect::StringProperty,
    has_pairing_delegate: inspect::UintProperty,
    evicted_peers: BoundedListNode,
    discovery_sessions: inspect::UintProperty,
    discovery_history: BoundedListNode,
}

impl HostDispatcherInspect {
    pub fn new(inspect: inspect::Node) -> HostDispatcherInspect {
        HostDispatcherInspect {
            host_count: inspect.create_uint("host_count", 0),
            peer_count: inspect.create_uint("peer_count", 0),
            device_class: inspect.create_string("device_class", "default"),
            input_capability: inspect.create_string("input_capability", "unknown"),
            output_capability: inspect.create_string("output_capability", "unknown"),
            has_pairing_delegate: inspect.create_uint("has_pairing_delegate", 0),
            peers: inspect.create_child("peers"),
            hosts: inspect.create_child("hosts"),
            discovery_sessions: inspect.create_uint("discovery_sessions", 0),
            discovery_history: BoundedListNode::new(
                inspect.create_child("discovery_history"),
                RECENT_DISCOVERY_SESSIONS_COUNT,
            ),
            evicted_peers: BoundedListNode::new(
                inspect.create_child("recently_removed"),
                RECENTLY_REMOVED_PEERS_COUNT,
            ),
            _inspect: inspect,
        }
    }

    pub fn peers(&self) -> &inspect::Node {
        &self.peers
    }

    pub fn hosts(&self) -> &inspect::Node {
        &self.hosts
    }
}

/// The HostDispatcher acts as a proxy aggregating multiple HostAdapters
/// It appears as a Host to higher level systems, and is responsible for
/// routing commands to the appropriate HostAdapter
struct HostDispatcherState {
    host_devices: HashMap<HostId, HostDevice>,
    active_id: Option<HostId>,

    // Component storage.
    stash: Stash,

    // GAP state
    // Name, if set. If not set, hosts will not have a set name.
    name: Option<String>,
    appearance: Appearance,
    discovery: DiscoveryState,
    discoverable: Option<Weak<HostDiscoverableSession>>,
    config_settings: build_config::Config,
    peers: HashMap<PeerId, Inspectable<Peer>>,

    // Sender end of a futures::mpsc channel to send LocalServiceRequests to Generic Access Service.
    // When a new host adapter is recognized, we create a new GasProxy, which takes GAS requests
    // from the new host and forwards them along a clone of this channel to GAS
    gas_channel_sender: mpsc::Sender<LocalServiceRequest>,

    pairing_dispatcher: Option<PairingDispatcherHandle>,

    watch_peers_publisher: hanging_get::Publisher<HashMap<PeerId, Peer>>,
    watch_peers_registrar: hanging_get::SubscriptionRegistrar<PeerWatcher>,

    watch_hosts_publisher: hanging_get::Publisher<Vec<HostInfo>>,
    watch_hosts_registrar: hanging_get::SubscriptionRegistrar<sys::HostWatcherWatchResponder>,

    // Pending requests to obtain a Host.
    host_requests: Slab<Waker>,

    inspect: HostDispatcherInspect,
}

impl HostDispatcherState {
    /// Set the active adapter for this HostDispatcher
    pub fn set_active_host(&mut self, adapter_id: HostId) -> types::Result<()> {
        if let Some(id) = self.active_id {
            if id == adapter_id {
                return Ok(());
            }

            // Shut down the previously active host.
            let _ = self.host_devices[&id].shutdown();
        }

        if self.host_devices.contains_key(&adapter_id) {
            self.set_active_id(Some(adapter_id));
            Ok(())
        } else {
            Err(types::Error::no_host())
        }
    }

    /// Used to set the pairing delegate. If there is a prior pairing delegate connected to the
    /// host, check if the existing stored connection is closed:
    ///  * if it is closed, overwrite it and succeed
    ///  * if it is still active, fail
    /// If there is no prior delegate, this will always succeed
    /// Returns `true` if the delegate was set successfully, otherwise false
    fn set_pairing_delegate(
        &mut self,
        delegate: PairingDelegateProxy,
        input: InputCapability,
        output: OutputCapability,
    ) -> types::Result<()> {
        match self.pairing_dispatcher.as_ref() {
            Some(dispatcher) if !dispatcher.is_closed() => {
                Err(format_err!("Another Delegate is active"))?
            }
            _ => {
                self.inspect.input_capability.set(&input.debug());
                self.inspect.output_capability.set(&output.debug());
                self.inspect.has_pairing_delegate.set(true.to_property());
                let (dispatcher, handle) = PairingDispatcher::new(delegate, input, output);
                for host in self.host_devices.values() {
                    handle.add_host(host.id(), host.proxy().clone());
                }
                // Old pairing dispatcher dropped; this drops all host pairings
                self.pairing_dispatcher = Some(handle);
                // Spawn handling of the new pairing requests
                // TODO(https://fxbug.dev/42152480) - We should avoid detach() here, and consider a more
                // explicit way to track this task
                fasync::Task::spawn(dispatcher.run()).detach();
                Ok(())
            }
        }
    }

    /// Return the active id. If the ID is currently not set,
    /// it will make the first ID in it's host_devices active
    fn get_active_id(&mut self) -> Option<HostId> {
        let active = self.active_id.clone();
        active.or_else(|| {
            self.host_devices.keys().next().cloned().map(|id| {
                self.set_active_id(Some(id));
                id
            })
        })
    }

    /// Return the active host. If the Host is currently not set,
    /// it will make the first ID in it's host_devices active
    fn get_active_host(&mut self) -> Option<HostDevice> {
        self.get_active_id().and_then(|id| self.host_devices.get(&id)).cloned()
    }

    /// Resolves all pending OnAdapterFuture's. Called when we leave the init period (by seeing the
    /// first host device or when the init timer expires).
    fn resolve_host_requests(&mut self) {
        for waker in &self.host_requests {
            waker.1.wake_by_ref();
        }
    }

    fn add_host(&mut self, id: HostId, host: HostDevice) {
        if self.host_devices.insert(id, host.clone()).is_some() {
            warn!("Host replaced: {}", id.to_string())
        } else {
            info!("Host added: {}", id.to_string());
        }

        // If this is the only host, mark it as active.
        let _ = self.get_active_id();

        // Update inspect state
        self.inspect.host_count.set(self.host_devices.len() as u64);

        // Notify HostWatcher interface clients about the new device.
        self.notify_host_watchers();

        // Resolve pending adapter futures.
        self.resolve_host_requests();
    }

    /// Updates the active adapter and notifies listeners & host watchers.
    fn set_active_id(&mut self, id: Option<HostId>) {
        info!("New active adapter: {}", id.map_or("<none>".to_string(), |id| id.to_string()));
        self.active_id = id;
        self.notify_host_watchers();
    }

    pub fn notify_host_watchers(&self) {
        // The HostInfo::active field for the active host must be filled in later.
        let active_id = self.active_id;

        // Wait for the hanging get watcher to update so we can linearize updates
        let current_hosts: Vec<HostInfo> = self
            .host_devices
            .values()
            .map(|host| {
                let mut info = host.info();
                // Fill in HostInfo::active
                if let Some(active_id) = active_id {
                    info.active = active_id == host.id();
                }
                info
            })
            .collect();
        let mut publisher = self.watch_hosts_publisher.clone();
        fasync::Task::spawn(async move {
            publisher
                .set(current_hosts)
                .await
                .expect("Fatal error: Host Watcher HangingGet unreachable");
        })
        .detach();
    }
}

#[derive(Clone)]
pub struct HostDispatcher {
    state: Arc<RwLock<HostDispatcherState>>,
}

impl HostDispatcher {
    /// The HostDispatcher will forward all Generic Access Service requests to the mpsc::Receiver
    /// end of |gas_channel_sender|. It is the responsibility of this function's caller to ensure
    /// that these requests are handled. This can be done by passing the mpsc::Receiver into a
    /// GenericAccessService struct and ensuring its run method is scheduled.
    pub fn new(
        appearance: Appearance,
        stash: Stash,
        inspect: inspect::Node,
        gas_channel_sender: mpsc::Sender<LocalServiceRequest>,
        watch_peers_publisher: hanging_get::Publisher<HashMap<PeerId, Peer>>,
        watch_peers_registrar: hanging_get::SubscriptionRegistrar<PeerWatcher>,
        watch_hosts_publisher: hanging_get::Publisher<Vec<HostInfo>>,
        watch_hosts_registrar: hanging_get::SubscriptionRegistrar<sys::HostWatcherWatchResponder>,
    ) -> HostDispatcher {
        let hd = HostDispatcherState {
            active_id: None,
            host_devices: HashMap::new(),
            name: None,
            appearance,
            config_settings: build_config::load_default(),
            peers: HashMap::new(),
            gas_channel_sender,
            stash,
            discovery: DiscoveryState::NotDiscovering,
            discoverable: None,
            pairing_dispatcher: None,
            watch_peers_publisher,
            watch_peers_registrar,
            watch_hosts_publisher,
            watch_hosts_registrar,
            host_requests: Slab::new(),
            inspect: HostDispatcherInspect::new(inspect),
        };
        HostDispatcher { state: Arc::new(RwLock::new(hd)) }
    }

    pub fn when_hosts_found(&self) -> impl Future<Output = HostDispatcher> {
        WhenHostsFound::new(self.clone())
    }

    pub fn get_name(&self) -> String {
        self.state.read().name.clone().unwrap_or(DEFAULT_DEVICE_NAME.to_string())
    }

    pub fn get_appearance(&self) -> Appearance {
        self.state.read().appearance
    }

    pub async fn set_name(&self, name: String, replace: NameReplace) -> types::Result<()> {
        if NameReplace::Keep == replace && self.state.read().name.is_some() {
            return Ok(());
        }
        self.state.write().name = Some(name);
        match self.active_host().await {
            Some(host) => {
                let name = self.get_name();
                host.set_name(name).await
            }
            None => Err(types::Error::no_host()),
        }
    }

    pub async fn set_device_class(&self, class: DeviceClass) -> types::Result<()> {
        let class_repr = class.debug();
        let res = match self.active_host().await {
            Some(host) => host.set_device_class(class).await,
            None => Err(types::Error::no_host()),
        };

        // Update Inspect state
        if res.is_ok() {
            self.state.read().inspect.device_class.set(&class_repr);
        }
        res
    }

    /// Set the active adapter for this HostDispatcher
    pub fn set_active_host(&self, host: HostId) -> types::Result<()> {
        self.state.write().set_active_host(host)
    }

    /// Used to set the pairing delegate. If there is a prior pairing delegate connected to the
    /// host, check if the existing stored connection is closed:
    ///  * if it is closed, overwrite it and succeed
    ///  * if it is still active, fail
    /// If there is no prior delegate, this will always succeed
    /// Returns `true` if the delegate was set successfully, otherwise false
    pub fn set_pairing_delegate(
        &self,
        delegate: PairingDelegateProxy,
        input: InputCapability,
        output: OutputCapability,
    ) -> types::Result<()> {
        self.state.write().set_pairing_delegate(delegate, input, output)
    }

    pub async fn apply_sys_settings(&self, new_settings: sys::Settings) -> build_config::Config {
        let (host_devices, new_config) = {
            let mut state = self.state.write();
            state.config_settings = state.config_settings.update_with_sys_settings(&new_settings);
            (state.host_devices.clone(), state.config_settings.clone())
        };
        for (host_id, device) in host_devices {
            let fut = device.apply_sys_settings(&new_settings);
            if let Err(e) = fut.await {
                warn!("Unable to apply new settings to host {}: {:?}", host_id, e);
                let failed_host_path = device.path();
                self.rm_device(failed_host_path).await;
            }
        }
        new_config
    }

    async fn discover_on_active_host(&self) -> types::Result<DiscoverySessionProxy> {
        match self.active_host().await {
            Some(host) => HostDevice::start_discovery(&host),
            None => Err(types::Error::no_host()),
        }
    }

    fn make_discovery_on_closed_task(&self, proxy: DiscoverySessionProxy) -> fasync::Task<()> {
        fasync::Task::spawn(self.clone().process_discovery_on_closed(proxy))
    }

    async fn process_discovery_on_closed(self, proxy: DiscoverySessionProxy) {
        // wait for DiscoverySession to close
        let _ = proxy.on_closed().await;
        debug!(
            "process_discovery_event_stream: Discovery protocol closed (state: {:?})",
            self.state.read().discovery
        );

        let old_state =
            std::mem::replace(&mut self.state.write().discovery, DiscoveryState::NotDiscovering);
        match old_state {
            DiscoveryState::NotDiscovering | DiscoveryState::Pending { .. } => {
                warn!("process_discovery_event_stream: Unexpected discovery event stream close in state {:?}", old_state);
            }
            DiscoveryState::Discovering { discovery_stopped_sender, .. } => {
                let _ = discovery_stopped_sender.send(());
            }
            DiscoveryState::Stopping { discovery_stopped_sender, session_sender, .. } => {
                let _ = discovery_stopped_sender.send(());

                // Restart discovery if clients queued session watchers while stopping.
                if let Some(sender) = session_sender {
                    // On errors all session_watchers will be dropped, signaling receivers of
                    // the error.
                    if let Ok(session) = self.start_discovery().await {
                        let _ = sender.send(session.clone());
                    }
                }
            }
        };
        trace!("discovery_event_stream_task: completed");
    }

    fn make_start_discovery_task(
        &self,
        session: Arc<DiscoverySession>,
        started: fasync::Time,
    ) -> fasync::Task<()> {
        let hd = self.clone();
        fasync::Task::spawn(async move {
            debug!("start_discovery_task: waiting for discover_on_active_host");
            let Ok(discovery_proxy) = hd.discover_on_active_host().await else {
                // On failure (host init timeout), revert state to NotDiscovering and drop Pending
                // state to notify Receivers.
                hd.state.write().discovery = DiscoveryState::NotDiscovering;
                return;
            };
            debug!("start_discovery_task: started discovery successfully");
            let _ = hd.state.read().inspect.discovery_sessions.add(1);

            let discovery_on_closed_task =
                hd.make_discovery_on_closed_task(discovery_proxy.clone());

            // Replace Pending state with new session and send session token to waiters
            let old_state =
                std::mem::replace(&mut hd.state.write().discovery, DiscoveryState::NotDiscovering);
            if let DiscoveryState::Pending {
                session_receiver: _,
                session_sender,
                discovery_stopped_receiver,
                discovery_stopped_sender,
                start_discovery_task: _,
            } = old_state
            {
                hd.state.write().discovery = DiscoveryState::Discovering {
                    session: Arc::downgrade(&session),
                    discovery_proxy,
                    started,
                    discovery_on_closed_task,
                    discovery_stopped_receiver,
                    discovery_stopped_sender,
                };
                let _ = session_sender.send(session.clone());
            }
            trace!("start_discovery_task: completed");
        })
    }

    pub async fn start_discovery(&self) -> types::Result<Arc<DiscoverySession>> {
        let session_fut = self.state.write().discovery.get_active_session_future();
        if let Some(session_fut) = session_fut {
            debug!(
                "start_discovery: awaiting DiscoverySession (state: {:?})",
                self.state.read().discovery
            );
            return session_fut
                .await
                .map_err(|_| format_err!("Pending discovery client channel closed").into());
        }

        let session = Arc::new(DiscoverySession { dispatcher_state: self.state.clone() });
        let started = fasync::Time::now();
        let (session_sender, session_receiver) = oneshot::channel();
        let session_receiver = session_receiver.shared();
        let (discovery_stopped_sender, discovery_stopped_receiver) = oneshot::channel();
        let discovery_stopped_receiver = discovery_stopped_receiver.shared();
        let start_discovery_task = self.make_start_discovery_task(session, started);

        // Immediately mark the state as pending to indicate to other requests to wait on
        // this discovery session initialization
        self.state.write().discovery = DiscoveryState::Pending {
            session_receiver: session_receiver.clone(),
            session_sender,
            discovery_stopped_receiver: discovery_stopped_receiver.clone(),
            discovery_stopped_sender,
            start_discovery_task,
        };

        debug!(
            "start_discovery: awaiting DiscoverySession for first client (state: {:?})",
            self.state.read().discovery
        );
        session_receiver
            .await
            .map_err(|_| format_err!("Pending discovery client channel closed").into())
    }

    // TODO(https://fxbug.dev/42139629) - This is susceptible to the same ToCtoToU race condition as
    // start_discovery. We can fix with the same tri-state pattern as for discovery
    pub async fn set_discoverable(&self) -> types::Result<Arc<HostDiscoverableSession>> {
        let strong_current_token =
            self.state.read().discoverable.as_ref().and_then(|token| token.upgrade());
        if let Some(token) = strong_current_token {
            return Ok(Arc::clone(&token));
        }

        match self.active_host().await {
            Some(host) => {
                let token = Arc::new(host.establish_discoverable_session().await?);
                self.state.write().discoverable = Some(Arc::downgrade(&token));
                Ok(token)
            }
            None => Err(types::Error::no_host()),
        }
    }

    fn stash(&self) -> Stash {
        self.state.read().stash.clone()
    }

    pub async fn forget(&self, peer_id: PeerId) -> types::Result<()> {
        // Try to delete from each host, even if it might not have the peer.
        // peers will be updated by the disconnection(s).
        let hosts = self.get_all_adapters().await;
        if hosts.is_empty() {
            return Err(sys::Error::Failed.into());
        }
        let mut hosts_removed: u32 = 0;
        for host in hosts {
            let host_path = host.path();

            match host.forget(peer_id).await {
                Ok(()) => hosts_removed += 1,
                Err(types::Error::SysError(sys::Error::PeerNotFound)) => {
                    trace!("No peer {} on host {:?}; ignoring", peer_id, host_path);
                }
                err => {
                    error!("Could not forget peer {} on host {:?}", peer_id, host_path);
                    return err;
                }
            }
        }

        if let Err(_) = self.stash().rm_peer(peer_id).await {
            return Err(format_err!("Couldn't remove peer").into());
        }

        if hosts_removed == 0 {
            return Err(format_err!("No hosts had peer").into());
        }
        Ok(())
    }

    pub async fn connect(&self, peer_id: PeerId) -> types::Result<()> {
        let host = self.active_host().await;
        match host {
            Some(host) => host.connect(peer_id).await,
            None => Err(types::Error::SysError(sys::Error::Failed)),
        }
    }

    /// Instruct the active host to intitiate a pairing procedure with the target peer. If it
    /// fails, we return the error we receive from the host
    pub async fn pair(&self, id: PeerId, pairing_options: PairingOptions) -> types::Result<()> {
        let host = self.active_host().await;
        match host {
            Some(host) => host.pair(id, pairing_options.into()).await,
            None => Err(sys::Error::Failed.into()),
        }
    }

    // Attempt to disconnect peer with id `peer_id` from all transports
    pub async fn disconnect(&self, peer_id: PeerId) -> types::Result<()> {
        let host = self.active_host().await;
        match host {
            Some(host) => host.disconnect(peer_id).await,
            None => Err(types::Error::no_host()),
        }
    }

    pub fn active_host(&self) -> impl Future<Output = Option<HostDevice>> {
        self.when_hosts_found().map(|adapter| {
            let mut wstate = adapter.state.write();
            wstate.get_active_host()
        })
    }

    pub async fn get_all_adapters(&self) -> Vec<HostDevice> {
        let _ = self.when_hosts_found().await;
        self.state.read().host_devices.values().cloned().collect()
    }

    #[cfg(test)]
    pub fn get_adapters(&self) -> Vec<HostInfo> {
        let hosts = self.state.read();
        hosts.host_devices.values().map(|host| host.info()).collect()
    }

    pub async fn request_host_service(self, chan: zx::Channel, service: HostService) {
        match self.active_host().await {
            Some(host) => {
                let host = host.proxy();
                match service {
                    HostService::LeCentral => {
                        let remote = ServerEnd::<CentralMarker>::new(chan.into());
                        let _ = host.request_protocol(ProtocolRequest::Central(remote));
                    }
                    HostService::LePeripheral => {
                        let remote = ServerEnd::<PeripheralMarker>::new(chan.into());
                        let _ = host.request_protocol(ProtocolRequest::Peripheral(remote));
                    }
                    HostService::LeGatt => {
                        let remote = ServerEnd::<Server_Marker>::new(chan.into());
                        let _ = host.request_protocol(ProtocolRequest::GattServer(remote));
                    }
                    HostService::LeGatt2 => {
                        let remote = ServerEnd::<Server_Marker2>::new(chan.into());
                        let _ = host.request_protocol(ProtocolRequest::Gatt2Server(remote));
                    }
                    HostService::Profile => {
                        let remote = ServerEnd::<ProfileMarker>::new(chan.into());
                        let _ = host.request_protocol(ProtocolRequest::Profile(remote));
                    }
                }
            }
            None => eprintln!("Failed to spawn, no active host"),
        }
    }

    // This is not an async method as we do not want to borrow `self` for the duration of the async
    // call, and we also want to trigger the send immediately even if the future is not yet awaited
    pub fn store_bond(&self, bond_data: BondingData) -> impl Future<Output = Result<(), Error>> {
        self.stash().store_bond(bond_data)
    }

    pub fn on_device_updated(&self, peer: Peer) -> impl Future<Output = ()> {
        let update_peer = peer.clone();

        let mut publisher = {
            let mut state = self.state.write();

            let node = state.inspect.peers().create_child(unique_name("peer_"));
            let peer = Inspectable::new(peer, node);
            let _drop_old_value = state.peers.insert(peer.id.clone(), peer);
            state.inspect.peer_count.set(state.peers.len() as u64);
            state.watch_peers_publisher.clone()
        };

        // Wait for the hanging get watcher to update so we can linearize updates
        async move {
            publisher
                .update(move |peers| {
                    let _ = peers.insert(update_peer.id, update_peer);
                    true
                })
                .await
                .expect("Fatal error: Peer Watcher HangingGet unreachable")
        }
    }

    pub fn on_device_removed(&self, id: PeerId) -> impl Future<Output = ()> {
        let mut publisher = {
            let mut state = self.state.write();
            if let Some(removed) = state.peers.remove(&id) {
                inspect_log!(state.inspect.evicted_peers, peer: removed);
            }
            state.inspect.peer_count.set(state.peers.len() as u64);
            state.watch_peers_publisher.clone()
        };

        // Wait for the hanging get watcher to update so we can linearize updates
        async move {
            publisher
                .update(move |peers| {
                    // Updated if we actually removed something.
                    peers.remove(&id).is_some()
                })
                .await
                .expect("Fatal error: Peer Watcher HangingGet unreachable")
        }
    }

    pub async fn watch_peers(&self) -> hanging_get::Subscriber<PeerWatcher> {
        let mut registrar = self.state.write().watch_peers_registrar.clone();
        registrar.new_subscriber().await.expect("Fatal error: Peer Watcher HangingGet unreachable")
    }

    pub async fn watch_hosts(&self) -> hanging_get::Subscriber<sys::HostWatcherWatchResponder> {
        let mut registrar = self.state.write().watch_hosts_registrar.clone();
        registrar.new_subscriber().await.expect("Fatal error: Host Watcher HangingGet unreachable")
    }

    async fn spawn_gas_proxy(&self, gatt_server_proxy: Server_Proxy) -> Result<(), Error> {
        let gas_channel = self.state.read().gas_channel_sender.clone();
        let gas_proxy =
            generic_access_service::GasProxy::new(gatt_server_proxy, gas_channel).await?;
        fasync::Task::spawn(gas_proxy.run().map(|r| {
            r.unwrap_or_else(|err| {
                warn!("Error passing message through Generic Access proxy: {:?}", err);
            })
        }))
        .detach();
        Ok(())
    }

    /// Commit all bootstrapped bonding identities to the system. This will update both the Stash
    /// and our in memory store, and notify all hosts of new bonding identities. If we already have
    /// bonding data for any of the peers (as identified by address), the new bootstrapped data
    /// will override them.
    pub async fn commit_bootstrap(&self, identities: Vec<Identity>) -> types::Result<()> {
        // Store all new bonds in our permanent Store. If we cannot successfully record the bonds
        // in the store, then Bootstrap.Commit() has failed.
        let mut stash = self.state.read().stash.clone();
        for identity in identities {
            stash.store_bonds(identity.bonds).await?
        }

        // Notify all current hosts of any changes to their bonding data
        let host_devices: Vec<_> = self.state.read().host_devices.values().cloned().collect();

        for host in host_devices {
            // If we fail to restore bonds to a given host, that is not a failure on a part of
            // Bootstrap.Commit(), but a failure on the host. So do not return error from this
            // function, but instead log and continue.
            // TODO(https://fxbug.dev/42121837) - if a host fails we should close it and clean up after it
            if let Err(error) =
                try_restore_bonds(host.clone(), self.clone(), &host.public_address()).await
            {
                error!(
                    "Error restoring Bootstrapped bonds to host '{:?}': {}",
                    host.debug_identifiers(),
                    error
                )
            }
        }
        Ok(())
    }

    /// Finishes initializing a host device by setting host configs and services.
    async fn add_host_device(&self, host_device: &HostDevice) -> Result<(), Error> {
        let dbg_ids = host_device.debug_identifiers();

        // TODO(https://fxbug.dev/42145442): Make sure that the bt-host device is left in a well-known state if
        // any of these operations fails.

        let address = host_device.public_address();
        assign_host_data(host_device.clone(), self.clone(), &address)
            .await
            .context(format!("{:?}: failed to assign identity to bt-host", dbg_ids))?;
        try_restore_bonds(host_device.clone(), self.clone(), &address)
            .await
            .map_err(|e| e.as_failure())?;

        let config = self.state.read().config_settings.clone();
        host_device
            .apply_config(config)
            .await
            .context(format!("{:?}: failed to configure bt-host device", dbg_ids))?;

        // Assign the name that is currently assigned to the HostDispatcher as the local name.
        let name = self.get_name();
        host_device
            .set_name(name)
            .await
            .map_err(|e| e.as_failure())
            .context(format!("{:?}: failed to set name of bt-host", dbg_ids))?;

        let (gatt_server_proxy, remote_gatt_server) = fidl::endpoints::create_proxy()?;
        host_device
            .proxy()
            .request_protocol(ProtocolRequest::Gatt2Server(remote_gatt_server))
            .context(format!("{:?}: failed to open gatt server for bt-host", dbg_ids))?;
        self.spawn_gas_proxy(gatt_server_proxy)
            .await
            .context(format!("{:?}: failed to spawn generic access service", dbg_ids))?;

        // Ensure the current active pairing delegate (if it exists) handles this host
        self.handle_pairing_requests(host_device.clone());

        self.state.write().add_host(host_device.id(), host_device.clone());

        Ok(())
    }

    // Update our hanging_get server with the latest hosts. This will notify any pending
    // hanging_gets and any new requests will see the new results.
    fn notify_host_watchers(&self) {
        self.state.write().notify_host_watchers();
    }

    pub async fn rm_device(&self, host_path: &str) {
        let mut new_adapter_activated = false;
        // Scope our HostDispatcherState lock
        {
            let mut hd = self.state.write();
            let active_id = hd.active_id.clone();

            // Get the host IDs that match `host_path`.
            let ids: Vec<HostId> = hd
                .host_devices
                .iter()
                .filter(|(_, ref host)| host.path() == host_path)
                .map(|(k, _)| k.clone())
                .collect();

            let id_strs: Vec<String> = ids.iter().map(|id| id.to_string()).collect();
            info!("Host removed: {} (path: {:?})", id_strs.join(","), host_path);

            for id in &ids {
                drop(hd.host_devices.remove(id));
            }

            // Reset the active ID if it got removed.
            if let Some(active_id) = active_id {
                if ids.contains(&active_id) {
                    hd.active_id = None;
                }
            }

            // Try to assign a new active adapter. This may send an "OnActiveAdapterChanged" event.
            if hd.active_id.is_none() && hd.get_active_id().is_some() {
                new_adapter_activated = true;
            }
        } // Now the lock is dropped, we can run the async notify

        if new_adapter_activated {
            if let Err(err) = self.configure_newly_active_adapter().await {
                warn!("Failed to persist state on adapter change: {:?}", err);
            }
        }
        self.notify_host_watchers();
    }

    /// Configure a newly active adapter with the correct behavior for an active adapter.
    async fn configure_newly_active_adapter(&self) -> types::Result<()> {
        // Migrate discovery state to new host
        let old_state =
            std::mem::replace(&mut self.state.write().discovery, DiscoveryState::NotDiscovering);
        match old_state {
            DiscoveryState::NotDiscovering => {}
            DiscoveryState::Pending {
                session_receiver,
                session_sender,
                discovery_stopped_receiver,
                discovery_stopped_sender,
                start_discovery_task,
            } => {
                info!("migrating Pending discovery to new host");
                // Stop the old discovery task, and restart discovery on the new host.
                drop(start_discovery_task);
                let session = Arc::new(DiscoverySession { dispatcher_state: self.state.clone() });
                let started = fasync::Time::now();
                let start_discovery_task = self.make_start_discovery_task(session, started);
                self.state.write().discovery = DiscoveryState::Pending {
                    session_receiver,
                    session_sender,
                    discovery_stopped_receiver,
                    discovery_stopped_sender,
                    start_discovery_task,
                };
            }
            DiscoveryState::Discovering {
                session,
                discovery_proxy,
                started,
                discovery_on_closed_task,
                discovery_stopped_receiver,
                discovery_stopped_sender,
            } => {
                info!("migrating Discovering discovery to new host");
                drop(discovery_on_closed_task);
                drop(discovery_proxy);

                let session = session.upgrade().ok_or(format_err!("failed to upgrade session"))?;

                // Restart discovery.
                let (session_sender, session_receiver) = oneshot::channel();
                let session_receiver = session_receiver.shared();
                let start_discovery_task = self.make_start_discovery_task(session, started);
                self.state.write().discovery = DiscoveryState::Pending {
                    session_receiver: session_receiver.clone(),
                    session_sender,
                    discovery_stopped_receiver,
                    discovery_stopped_sender,
                    start_discovery_task,
                };
                return session_receiver
                    .await
                    .map(|_| ())
                    .map_err(|_| format_err!("Pending discovery client channel closed").into());
            }
            DiscoveryState::Stopping {
                discovery_proxy,
                session_receiver: _,
                session_sender,
                discovery_on_closed_task: discovery_event_stream_task,
                discovery_stopped_receiver: _,
                discovery_stopped_sender,
            } => {
                info!("migrating Stopping discovery to new host");
                drop(discovery_event_stream_task);
                drop(discovery_proxy);
                let _ = discovery_stopped_sender.send(());

                // Restart discovery if clients queued session receiver while stopping.
                if let Some(sender) = session_sender {
                    // On errors, sender will be dropped, signaling receivers of
                    // the error.
                    if let Ok(session) = self.start_discovery().await {
                        let _ = sender.send(session.clone());
                    }
                }
            }
        }

        Ok(())
    }

    /// Route pairing requests from this host through our pairing dispatcher, if it exists
    fn handle_pairing_requests(&self, host: HostDevice) {
        let mut dispatcher = self.state.write();

        if let Some(handle) = &mut dispatcher.pairing_dispatcher {
            handle.add_host(host.id(), host.proxy().clone());
        }
    }

    pub async fn add_host_component(&self, proxy: HostProxy) -> types::Result<()> {
        info!("Adding host component");

        let node = self.state.read().inspect.hosts().create_child(unique_name("device_"));

        let proxy_handle = proxy.as_channel().raw_handle().to_string();
        let host_device = init_host(proxy_handle.as_str(), node, proxy).await?;
        info!("Successfully started host device: {:?}", host_device.info());
        self.add_host_device(&host_device).await?;

        // Start listening to Host interface events.
        fasync::Task::spawn({
            let this = self.clone();
            async move {
                match host_device.watch_events(this.clone()).await {
                    Ok(()) => (),
                    Err(e) => {
                        warn!("Error handling host event: {e:?}");
                        let host_path = proxy_handle.as_str();
                        this.rm_device(&host_path).await;
                    }
                }
            }
        })
        .detach();

        Ok(())
    }
}

async fn init_host(path: &str, node: inspect::Node, proxy: HostProxy) -> Result<HostDevice, Error> {
    node.record_string("path", path);

    // Obtain basic information and create and entry in the dispatcher's map.
    let host_info = proxy.watch_state().await.context("failed to obtain bt-host information")?;
    let host_info = Inspectable::new(HostInfo::try_from(host_info)?, node);

    Ok(HostDevice::new(path.to_string(), proxy, host_info))
}

impl HostListener for HostDispatcher {
    type PeerUpdatedFut = BoxFuture<'static, ()>;
    fn on_peer_updated(&mut self, peer: Peer) -> Self::PeerUpdatedFut {
        self.on_device_updated(peer).boxed()
    }
    type PeerRemovedFut = BoxFuture<'static, ()>;
    fn on_peer_removed(&mut self, id: PeerId) -> Self::PeerRemovedFut {
        self.on_device_removed(id).boxed()
    }
    type HostBondFut = BoxFuture<'static, Result<(), anyhow::Error>>;
    fn on_new_host_bond(&mut self, data: BondingData) -> Self::HostBondFut {
        self.store_bond(data).boxed()
    }

    type HostInfoFut = BoxFuture<'static, Result<(), anyhow::Error>>;
    fn on_host_updated(&mut self, _info: HostInfo) -> Self::HostInfoFut {
        self.notify_host_watchers();
        async { Ok(()) }.boxed()
    }
}

/// A future that completes when at least one adapter is available.
#[must_use = "futures do nothing unless polled"]
struct WhenHostsFound {
    hd: HostDispatcher,
    waker_key: Option<usize>,
}

impl WhenHostsFound {
    // Constructs an WhenHostsFound that completes at the latest after HOST_INIT_TIMEOUT seconds.
    fn new(hd: HostDispatcher) -> impl Future<Output = HostDispatcher> {
        WhenHostsFound { hd: hd.clone(), waker_key: None }.on_timeout(
            Duration::from_seconds(HOST_INIT_TIMEOUT).after_now(),
            move || {
                {
                    let mut inner = hd.state.write();
                    if inner.host_devices.len() == 0 {
                        info!("No bt-host devices found");
                        inner.resolve_host_requests();
                    }
                }
                hd
            },
        )
    }

    fn remove_waker(&mut self) {
        if let Some(key) = self.waker_key {
            drop(self.hd.state.write().host_requests.remove(key));
        }
        self.waker_key = None;
    }
}

impl Drop for WhenHostsFound {
    fn drop(&mut self) {
        self.remove_waker()
    }
}

impl Unpin for WhenHostsFound {}

impl Future for WhenHostsFound {
    type Output = HostDispatcher;

    fn poll(mut self: ::std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.hd.state.read().host_devices.len() == 0 {
            let hd = self.hd.clone();
            if self.waker_key.is_none() {
                self.waker_key = Some(hd.state.write().host_requests.insert(cx.waker().clone()));
            }
            Poll::Pending
        } else {
            self.remove_waker();
            Poll::Ready(self.hd.clone())
        }
    }
}

async fn try_restore_bonds(
    host_device: HostDevice,
    hd: HostDispatcher,
    address: &Address,
) -> types::Result<()> {
    // Load bonding data that use this host's `address` as their "local identity address".
    let opt_data = hd.stash().list_bonds(address.clone()).await?;
    let data = match opt_data {
        Some(data) => data,
        None => return Ok(()),
    };
    match host_device.restore_bonds(data).await {
        Err(e) => {
            error!("failed to restore bonding data for host: {:?}", e);
            Err(e)
        }
        Ok(errors) => {
            if errors.is_empty() {
                Ok(())
            } else {
                let msg =
                    errors.into_iter().fold("".to_string(), |acc, b| format!("{}, {:?}", acc, b));
                let msg = format!("failed to restore bonding data: {}", msg);
                error!("{}", msg);
                Err(anyhow!(msg).into())
            }
        }
    }
}

fn generate_irk() -> Result<sys::Key, zx::Status> {
    let mut buf: [u8; 16] = [0; 16];
    // Generate a secure IRK.
    zx::cprng_draw(&mut buf);
    Ok(sys::Key { value: buf })
}

async fn assign_host_data(
    host: HostDevice,
    hd: HostDispatcher,
    address: &Address,
) -> Result<(), Error> {
    // Obtain an existing IRK or generate a new one if one doesn't already exists for |address|.
    let data = match hd.stash().get_host_data(address.clone()).await? {
        Some(host_data) => {
            trace!("restored IRK");
            host_data.clone()
        }
        None => {
            // Generate a new IRK.
            trace!("generating new IRK");
            let new_data = HostData { irk: Some(generate_irk()?) };

            if let Err(e) = hd.stash().store_host_data(address.clone(), new_data.clone()).await {
                error!("failed to persist local IRK");
                return Err(e.into());
            }
            new_data
        }
    };
    host.set_local_data(data).map_err(|e| e.into())
}

#[cfg(test)]
pub(crate) mod test {
    use super::*;

    use fidl_fuchsia_bluetooth_gatt2::{
        LocalServiceProxy, Server_Request, Server_RequestStream as GattServerRequestStream,
    };
    use fidl_fuchsia_bluetooth_host::{HostRequest, HostRequestStream};
    use futures::{future::join, StreamExt};

    pub(crate) fn make_test_dispatcher(
        watch_peers_publisher: hanging_get::Publisher<HashMap<PeerId, Peer>>,
        watch_peers_registrar: hanging_get::SubscriptionRegistrar<PeerWatcher>,
        watch_hosts_publisher: hanging_get::Publisher<Vec<HostInfo>>,
        watch_hosts_registrar: hanging_get::SubscriptionRegistrar<sys::HostWatcherWatchResponder>,
    ) -> HostDispatcher {
        let (gas_channel_sender, _ignored_gas_task_req_stream) = mpsc::channel(0);
        HostDispatcher::new(
            Appearance::Display,
            Stash::in_memory_mock(),
            fuchsia_inspect::Node::default(),
            gas_channel_sender,
            watch_peers_publisher,
            watch_peers_registrar,
            watch_hosts_publisher,
            watch_hosts_registrar,
        )
    }

    pub(crate) fn make_simple_test_dispatcher() -> HostDispatcher {
        let watch_peers_broker = hanging_get::HangingGetBroker::new(
            HashMap::new(),
            |_, _| true,
            hanging_get::DEFAULT_CHANNEL_SIZE,
        );
        let watch_hosts_broker = hanging_get::HangingGetBroker::new(
            Vec::new(),
            |_, _| true,
            hanging_get::DEFAULT_CHANNEL_SIZE,
        );

        let dispatcher = make_test_dispatcher(
            watch_peers_broker.new_publisher(),
            watch_peers_broker.new_registrar(),
            watch_hosts_broker.new_publisher(),
            watch_hosts_broker.new_registrar(),
        );

        let watchers_fut = join(watch_peers_broker.run(), watch_hosts_broker.run()).map(|_| ());
        fasync::Task::spawn(watchers_fut).detach();
        dispatcher
    }

    #[derive(Default)]
    pub(crate) struct GasEndpoints {
        gatt_server: Option<GattServerRequestStream>,
        service: Option<LocalServiceProxy>,
    }

    async fn handle_standard_host_server_init(
        mut host_server: HostRequestStream,
    ) -> (HostRequestStream, GasEndpoints) {
        let mut gas_endpoints = GasEndpoints::default();
        while gas_endpoints.gatt_server.is_none() {
            match host_server.next().await {
                Some(Ok(HostRequest::SetLocalName { responder, .. })) => {
                    info!("Setting Local Name");
                    let _ = responder.send(Ok(()));
                }
                Some(Ok(HostRequest::SetDeviceClass { responder, .. })) => {
                    info!("Setting Device Class");
                    let _ = responder.send(Ok(()));
                }
                Some(Ok(HostRequest::RequestProtocol {
                    payload: ProtocolRequest::Gatt2Server(server),
                    ..
                })) => {
                    // don't respond at all on the server side.
                    info!("Storing Gatt Server");
                    let mut gatt_server = server.into_stream().unwrap();
                    info!("GAS Server was started, waiting for publish");
                    // The Generic Access Service now publishes itself.
                    match gatt_server.next().await {
                        Some(Ok(Server_Request::PublishService { info, service, responder })) => {
                            info!("Captured publish of GAS Service: {:?}", info);
                            gas_endpoints.service = Some(service.into_proxy().unwrap());
                            let _ = responder.send(Ok(()));
                        }
                        x => error!("Got unexpected GAS Server request: {:?}", x),
                    }
                    gas_endpoints.gatt_server = Some(gatt_server);
                }
                Some(Ok(HostRequest::SetConnectable { responder, .. })) => {
                    info!("Setting connectable");
                    let _ = responder.send(Ok(()));
                }
                Some(Ok(req)) => info!("Unhandled Host Request in add: {:?}", req),
                Some(Err(e)) => error!("Error in host server: {:?}", e),
                None => break,
            }
        }
        info!("Finishing host_device mocking for add host");
        (host_server, gas_endpoints)
    }

    pub(crate) async fn create_and_add_test_host_to_dispatcher(
        id: HostId,
        dispatcher: &HostDispatcher,
    ) -> types::Result<(HostRequestStream, HostDevice, GasEndpoints)> {
        let (host_server, host_device) = HostDevice::mock_from_id(id);
        let host_server_init_handler = handle_standard_host_server_init(host_server);
        let (res, (host_server, gas_endpoints)) =
            join(dispatcher.add_host_device(&host_device), host_server_init_handler).await;
        res?;
        Ok((host_server, host_device, gas_endpoints))
    }
}