oc2_hlapi_v2 0.2.0

//! HLAPI high-level client

use std::{
    collections::{HashMap, VecDeque},
    error::Error,
    fmt::Debug,
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};

use erased_serde::Serialize as ErasedSerialize;
use serde::de::DeserializeOwned;
use serde_json::{Value as JsonValue, value::Serializer as JsonValueSerializer};
use thiserror::Error;
use uuid::Uuid;

use crate::{
    Codec,
    messages::{
        self, BlobDescriptor, DeviceInfo, Event, EventData, MethodInfo, MethodInvocation, Request,
        RequestData, Response, ResponseData,
    },
};
#[cfg(target_os = "linux")]
use crate::{
    codecs::BaseCodec,
    transports::{BusdClient, BusdUnixSocket, LinuxVirtio},
};

#[derive(Debug, Error)]
#[non_exhaustive]
/// Errors produced by the HLAPI client
pub enum HlapiClientError {
    /// Connection error
    #[error("Connection error: {0}")]
    Connection(Box<dyn Error + Send + Sync>),
    /// Codec error
    #[error("Codec error: {0}")]
    Codec(#[source] Box<dyn Error + Send + Sync>),
    /// HLAPI error
    #[error("HLAPI error: {0}")]
    Hlapi(String),
    /// Serde error
    #[error("Serde error: {0}")]
    Serde(#[from] serde_json::Error),
    #[error("Unexpected response type to HLAPI message")]
    /// Unexpected HLAPI response type
    UnexpectedResponseType,
    /// HLAPI client state mutex poisoned
    #[error("HLAPI client state mutex poisoned")]
    MutexPoisoned,
    #[error("Blob too big")]
    /// Blob too big
    BlobTooBig,
    #[error("Request timeout")]
    /// Request timeout
    RequestTimeout,
}
impl From<messages::BlobTooBig> for HlapiClientError {
    fn from(_value: messages::BlobTooBig) -> Self {
        Self::BlobTooBig
    }
}

trait ErrorBoxedCodec {
    fn send_request(
        &mut self,
        request: Request,
        blobs: &[&[u8]],
    ) -> Result<(), Box<dyn Error + Send + Sync>>;
    fn read_response(
        &mut self,
        timeout: Option<Duration>,
    ) -> Result<Option<Response>, Box<dyn Error + Send + Sync>>;
    fn read_event(
        &mut self,
        timeout: Option<Duration>,
    ) -> Result<Option<Event>, Box<dyn Error + Send + Sync>>;
    fn read_blob(
        &mut self,
        descriptor: BlobDescriptor,
    ) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>>;
    fn clear_response_stream(&mut self) -> Result<(), Box<dyn Error + Send + Sync>>;
    fn clear_blob_stream(&mut self) -> Result<(), Box<dyn Error + Send + Sync>>;
}

impl<T: Codec> ErrorBoxedCodec for T {
    fn send_request(
        &mut self,
        request: Request,
        blobs: &[&[u8]],
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        self.send_request(request, blobs)
            .map_err(|e| Box::new(e) as _)
    }

    fn read_response(
        &mut self,
        timeout: Option<Duration>,
    ) -> Result<Option<Response>, Box<dyn Error + Send + Sync>> {
        self.read_response(timeout).map_err(|e| Box::new(e) as _)
    }

    fn read_event(
        &mut self,
        timeout: Option<Duration>,
    ) -> Result<Option<Event>, Box<dyn Error + Send + Sync>> {
        self.read_event(timeout).map_err(|e| Box::new(e) as _)
    }

    fn read_blob(
        &mut self,
        descriptor: BlobDescriptor,
    ) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>> {
        self.read_blob(descriptor).map_err(|e| Box::new(e) as _)
    }

    fn clear_response_stream(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
        self.clear_response_stream().map_err(|e| Box::new(e) as _)
    }

    fn clear_blob_stream(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
        self.clear_blob_stream().map_err(|e| Box::new(e) as _)
    }
}

type EventListener =
    Box<dyn Fn(&str, &JsonValue) -> Result<(), Box<dyn Error + Send + Sync>> + Send + Sync>;

struct HlapiClientInner {
    codec: Box<dyn ErrorBoxedCodec + Send>,
    generation: Option<u32>,
    list_cache: Option<Vec<DeviceInfo>>,
    cache_maybe_invalid: bool,
    request_timeout: Option<Duration>,
    next_request_id: u32,
    event_listeners: HashMap<Uuid, HashMap<Uuid, EventListener>>,
}

impl HlapiClientInner {
    fn new(codec: Box<dyn ErrorBoxedCodec + Send>) -> Self {
        Self {
            codec,
            generation: None,
            list_cache: None,
            cache_maybe_invalid: false,
            request_timeout: Some(HlapiClient::DEFAULT_REQUEST_TIMEOUT),
            next_request_id: 1,
            event_listeners: HashMap::new(),
        }
    }

    fn get_list_cache(&mut self) -> Result<&Vec<DeviceInfo>, HlapiClientError> {
        if self.cache_maybe_invalid || self.list_cache.is_none() {
            let Response {
                data: ResponseData::List { device_list },
                ..
            } = self.make_request(
                Request {
                    id: None,
                    data: RequestData::List,
                },
                &[],
            )?
            else {
                return Err(HlapiClientError::UnexpectedResponseType);
            };
            self.list_cache = Some(device_list);
        }
        Ok(self.list_cache.as_ref().unwrap())
    }

    fn list_devices(&mut self) -> Result<Vec<DeviceInfo>, HlapiClientError> {
        self.process_events(Some(Duration::ZERO))?;
        Ok(self.get_list_cache()?.clone())
    }

    fn find_devices_by_names(
        &mut self,
        names: &[&str],
        device_type: &str,
    ) -> Result<Vec<Uuid>, HlapiClientError> {
        self.process_events(Some(Duration::ZERO))?;
        let list_cache = self.get_list_cache()?;
        Ok(list_cache
            .iter()
            .filter(|d| {
                names.iter().all(|&name| {
                    name.is_empty() || d.type_names.iter().any(|dev_name| dev_name == name)
                }) && (device_type.is_empty()
                    || d.type_names.iter().any(|dev_name| dev_name == device_type))
            })
            .map(|d| d.device_id)
            .collect::<Vec<_>>())
    }

    fn device_id_valid_for(
        &mut self,
        device_id: Uuid,
        typ: &str,
    ) -> Result<bool, HlapiClientError> {
        self.process_events(Some(Duration::ZERO))?;
        let list_cache = self.get_list_cache()?;
        Ok(list_cache.iter().any(|d| {
            d.device_id == device_id && (typ.is_empty() || d.type_names.iter().any(|x| x == typ))
        }))
    }

    fn make_request(
        &mut self,
        mut request: Request,
        blobs: &[&[u8]],
    ) -> Result<Response, HlapiClientError> {
        let id = self.next_request_id;
        self.next_request_id = self.next_request_id.wrapping_add(1);
        // Request IDs are signed in the Java code but should only be positive, so make sure we
        // never send a value greater than the signed integer max.
        if self.next_request_id > i32::MAX as u32 {
            self.next_request_id = 1;
        }
        // Zero is used as "no ID" in several places, so make sure it is skipped
        if self.next_request_id == 0 {
            self.next_request_id = 1;
        }
        request.id = Some(id);
        self.codec
            .clear_response_stream()
            .map_err(HlapiClientError::Codec)?;
        self.codec
            .clear_blob_stream()
            .map_err(HlapiClientError::Codec)?;
        self.codec
            .send_request(request, blobs)
            .map_err(HlapiClientError::Codec)?;

        let deadline = self.request_timeout.map(|t| Instant::now() + t);
        let resp = loop {
            let remaining = deadline.map(|d| d.saturating_duration_since(Instant::now()));
            let Some(resp) = self
                .codec
                .read_response(remaining)
                .map_err(HlapiClientError::Codec)?
            else {
                return Err(HlapiClientError::RequestTimeout);
            };
            // A non-zero request ID that isn't the ID we sent is a response to some other request we
            // timed out on reading the reply of, ignore it.
            if resp.id != 0 && resp.id != id {
                continue;
            }
            break resp;
        };
        if let Some(generation) = resp.generation {
            self.set_generation(generation);
        }
        if let ResponseData::Error { msg } = resp.data {
            return Err(HlapiClientError::Hlapi(msg));
        }
        Ok(resp)
    }

    #[allow(clippy::type_complexity)]
    fn call<R: DeserializeOwned>(
        &mut self,
        device_id: Uuid,
        method: &str,
        args: &[&dyn ErasedSerialize],
        blobs: &[&[u8]],
    ) -> Result<(R, Vec<Vec<u8>>), HlapiClientError> {
        self.process_events(Some(Duration::ZERO))?;
        let mut blob_descriptors = Vec::new();
        for &blob in blobs {
            blob_descriptors.push(BlobDescriptor::describe(blob)?);
        }
        let mut parameters = Vec::new();
        for &arg in args {
            parameters.push(erased_serde::serialize(&arg, JsonValueSerializer)?);
        }

        let Response {
            data:
                ResponseData::Result {
                    return_value,
                    blobs: return_blob_descriptors,
                },
            ..
        } = self.make_request(
            Request {
                id: None,
                data: RequestData::Invoke {
                    invocation: MethodInvocation {
                        device_id,
                        name: method.to_owned(),
                        parameters: Some(parameters),
                    },
                    blobs: Some(blob_descriptors),
                },
            },
            blobs,
        )?
        else {
            return Err(HlapiClientError::UnexpectedResponseType);
        };

        let mut return_blobs = Vec::new();
        for descriptor in return_blob_descriptors.unwrap_or(Vec::new()) {
            let blob = self
                .codec
                .read_blob(descriptor)
                .map_err(HlapiClientError::Codec)?;
            return_blobs.push(blob);
        }
        let return_value = return_value.unwrap_or(JsonValue::Null);
        let return_value = serde_json::from_value(return_value)?;
        Ok((return_value, return_blobs))
    }

    fn process_events(&mut self, timeout: Option<Duration>) -> Result<(), HlapiClientError> {
        if self.process_event(timeout)? {
            while self.process_event(Some(Duration::ZERO))? {}
        }
        Ok(())
    }

    fn process_event(&mut self, timeout: Option<Duration>) -> Result<bool, HlapiClientError> {
        let deadline = timeout.map(|x| Instant::now() + x);
        if let Some(event) = self
            .codec
            .read_event(deadline.map(|d| d.saturating_duration_since(Instant::now())))
            .map_err(HlapiClientError::Codec)?
        {
            self.set_generation(event.generation);
            if let EventData::EventsDropped { .. } = event.data {
                self.cache_maybe_invalid = true;
            }
            if let EventData::Device(device_event) = event.data
                && let Some(device_listener_map) =
                    self.event_listeners.get_mut(&device_event.device_id)
            {
                for listener in device_listener_map.values() {
                    let _ = listener(
                        &device_event.typ,
                        device_event.data.as_ref().unwrap_or(&JsonValue::Null),
                    );
                }
            }
            Ok(true)
        } else {
            Ok(false)
        }
    }

    fn register_event_listener(&mut self, device_id: Uuid, listener: EventListener) -> Uuid {
        let mut listener_uuid = Uuid::new_v4();
        let device_listener_map = self.event_listeners.entry(device_id).or_default();
        while device_listener_map.contains_key(&listener_uuid) {
            listener_uuid = Uuid::new_v4();
        }
        device_listener_map.insert(listener_uuid, listener);
        listener_uuid
    }

    fn unregister_event_listener(&mut self, device_id: Uuid, listener_id: Uuid) {
        if let Some(device_listener_map) = self.event_listeners.get_mut(&device_id) {
            device_listener_map.remove(&listener_id);
            if device_listener_map.is_empty() {
                self.event_listeners.remove(&device_id);
            }
        }
    }

    fn set_generation(&mut self, generation: u32) {
        if self.generation != Some(generation) {
            self.list_cache = None;
        }
        self.generation = Some(generation);
        self.cache_maybe_invalid = false;
    }
}

/// The HLAPI client
///
/// Cloned clients share state, including one internal connection, and an in-progress method on one
/// client will block the others. If you are calling a long-running method on a background thread,
/// it should have its own connection.
#[derive(Clone)]
pub struct HlapiClient {
    inner: Arc<Mutex<HlapiClientInner>>,
}

impl Debug for HlapiClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Since the mutex may be locked and the codec may not be Debug anyways, don't try to show
        // the client state in the Debug impl.
        f.debug_struct("HlapiClient").finish_non_exhaustive()
    }
}

#[cfg(target_os = "linux")]
impl HlapiClient {
    /// Helper function to connect to the HLAPI bus the way Lua does, trying the bus daemon first
    /// and falling back to virtio if that fails.
    pub fn connect() -> Result<Self, HlapiClientError> {
        if let Ok(client) = Self::connect_busd_client() {
            Ok(client)
        } else {
            Self::connect_virtio()
        }
    }

    /// Helper function to make a connection with the [`LinuxVirtio`] transport
    pub fn connect_virtio() -> Result<Self, HlapiClientError> {
        Ok(Self::new(
            BaseCodec::new(
                LinuxVirtio::new().map_err(|e| HlapiClientError::Connection(Box::new(e)))?,
            )
            .map_err(|e| HlapiClientError::Connection(Box::new(e)))?,
        ))
    }

    /// Helper function to make a connection with the [`BusdClient`] transport, autodetecting the
    /// socket path.
    pub fn connect_busd_client() -> Result<Self, HlapiClientError> {
        Ok(Self::new(
            BaseCodec::new(BusdClient::new(BusdUnixSocket::new()))
                .map_err(|e| HlapiClientError::Connection(Box::new(e)))?,
        ))
    }
}

impl HlapiClient {
    /// The default request timeout for HLAPI requests. 35s is the default for the Lua libraries, so we
    /// use it as well.
    pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(35);

    /// Create a new client with the given codec
    pub fn new<C: Codec + Send + 'static>(codec: C) -> Self {
        Self {
            inner: Arc::new(Mutex::new(HlapiClientInner::new(Box::new(codec)))),
        }
    }

    /// List all devices on the bus
    pub fn list_devices(&self) -> Result<Vec<DeviceInfo>, HlapiClientError> {
        self.inner
            .lock()
            .map_err(|_| HlapiClientError::MutexPoisoned)?
            .list_devices()
    }

    /// Find the first device with the type of the provided wrapper struct. If multiple devices
    /// exist, the one returned is the one with the lowest UUID.
    ///
    /// [`GenericDevice`] will match all device types.
    ///
    /// [`GenericDevice`]: crate::devices::GenericDevice
    pub fn find_device_by_type<D: DeviceBase>(&self) -> Result<Option<D>, HlapiClientError> {
        Ok(self.find_devices_by_type()?.into_iter().next())
    }

    /// Find the first device with all the given names, as well as the wrapper's device type. If multiple devices exist, the one returned
    /// is the one with the lowest UUID.
    ///
    /// An empty name matches all device names, and [`GenericDevice`] will match all device types.
    ///
    /// [`GenericDevice`]: crate::devices::GenericDevice
    pub fn find_device_by_names<D: DeviceBase>(
        &self,
        names: &[&str],
    ) -> Result<Option<D>, HlapiClientError> {
        Ok(self.find_devices_by_names(names)?.into_iter().next())
    }

    /// Find all devices with the type of the provided wrapper struct. The returned list is sorted by UUID.
    ///
    /// [`GenericDevice`] will match all device types.
    ///
    /// [`GenericDevice`]: crate::devices::GenericDevice
    pub fn find_devices_by_type<D: DeviceBase>(&self) -> Result<Vec<D>, HlapiClientError> {
        self.find_devices_by_names(&[])
    }

    /// Find all devices with all the given names, as well as the wrapper's device type. The returned list is sorted by UUID.
    ///
    /// An empty name matches all device names, and [`GenericDevice`] will match all device types.
    ///
    /// [`GenericDevice`]: crate::devices::GenericDevice
    pub fn find_devices_by_names<D: DeviceBase>(
        &self,
        names: &[&str],
    ) -> Result<Vec<D>, HlapiClientError> {
        let mut devices = self
            .inner
            .lock()
            .map_err(|_| HlapiClientError::MutexPoisoned)?
            .find_devices_by_names(names, D::TYPE)?;
        devices.sort();
        devices
            .iter()
            .map(|&x| D::new(self.clone(), x))
            .collect::<Result<_, _>>()
    }

    /// Get the device with the given UUID and type.
    ///
    /// [`GenericDevice`] will match all device types.
    ///
    /// [`GenericDevice`]: crate::devices::GenericDevice
    pub fn get_device_by_uuid<D: DeviceBase>(
        &self,
        uuid: Uuid,
    ) -> Result<Option<D>, HlapiClientError> {
        if self
            .inner
            .lock()
            .map_err(|_| HlapiClientError::MutexPoisoned)?
            .device_id_valid_for(uuid, D::TYPE)?
        {
            Ok(Some(D::new(self.clone(), uuid)?))
        } else {
            Ok(None)
        }
    }

    /// Set how long a request may take before it times out. `None` means no timeout.
    ///
    /// The default value is [`Self::DEFAULT_REQUEST_TIMEOUT`].
    pub fn set_request_timeout(&self, timeout: Option<Duration>) -> Result<(), HlapiClientError> {
        self.inner
            .lock()
            .map_err(|_| HlapiClientError::MutexPoisoned)?
            .request_timeout = timeout;
        Ok(())
    }

    fn make_request(&self, request: Request) -> Result<Response, HlapiClientError> {
        self.inner
            .lock()
            .map_err(|_| HlapiClientError::MutexPoisoned)?
            .make_request(request, &[])
    }

    #[allow(clippy::type_complexity)]
    fn call<R: DeserializeOwned>(
        &self,
        uuid: Uuid,
        method: &str,
        args: &[&dyn ErasedSerialize],
        blobs: &[&[u8]],
    ) -> Result<(R, Vec<Vec<u8>>), HlapiClientError> {
        self.inner
            .lock()
            .map_err(|_| HlapiClientError::MutexPoisoned)?
            .call(uuid, method, args, blobs)
    }
}

/// HLAPI base device trait
///
/// This is the trait for the base methods of a device type, allowing the HLAPI client to create and
/// get info about a device wrapper, and for device wrappers to call methods. Custom device wrappers
/// *must* implement this trait.
///
/// Users of this library should not use this trait unless they are
/// writing a custom device wrapper, the methods here are not part of the [`Device`] trait to separate
/// them for this exact reason. See the methods on [`HlapiClient`] to find devices, and the methods
/// on the [`Device`] trait for common methods for all devices.
pub trait DeviceBase {
    /// The device's type string
    const TYPE: &'static str;

    /// The event type for this device's events. Should be set to [`Infallible`] if [`parse_event`] is not implemented.
    ///
    /// [`Infallible`]: std::convert::Infallible
    /// [`parse_event`]: DeviceBase::parse_event
    type Event: 'static + Send;

    /// Create a device wrapper
    fn new(client: HlapiClient, uuid: Uuid) -> Result<Self, HlapiClientError>
    where
        Self: Sized;

    /// Gets the UUID of this device
    fn uuid_base(&self) -> Uuid;

    /// Gets the client associated with this device
    fn client(&self) -> &HlapiClient;

    /// Call a method on this device
    ///
    /// [`BlobMarker`] should be used to denote blobs in the method arguments and the return
    /// value.
    ///
    /// Do *not* override this method! It is the only way to perform method calls, as the client
    /// method that this uses to do so is a private method. If you override this you will have no
    /// way to make HLAPI method calls and thus cannot write a useful wrapper.
    ///
    /// [`BlobMarker`]: crate::messages::BlobMarker
    #[allow(clippy::type_complexity)]
    fn call<R: DeserializeOwned>(
        &self,
        method: &str,
        args: &[&dyn ErasedSerialize],
        blobs: &[&[u8]],
    ) -> Result<(R, Vec<Vec<u8>>), HlapiClientError> {
        self.client().call(self.uuid_base(), method, args, blobs)
    }

    /// Parses a device event from a type and a raw JSON value
    ///
    /// If this returns an error, the event is discarded. Currently nothing is done with the
    /// returned error value, though this may change in the future.
    fn parse_event(
        _typ: &str,
        _data: &JsonValue,
    ) -> Result<Self::Event, Box<dyn Error + Send + Sync>> {
        Err("device has no events".into())
    }
}

/// HLAPI device trait
///
/// This trait contains helper methods common to all devices. It is automatically implemented for
/// all wrappers, manually implementing it is not necessary or possible.
pub trait Device: DeviceBase {
    /// Gets the UUID of this device
    fn uuid(&self) -> Uuid;

    /// List the methods of the device
    fn list_methods(&self) -> Result<Vec<MethodInfo>, HlapiClientError>;

    /// Gets the stream of events from the device
    fn event_stream(&self) -> EventStream<Self::Event>;
}

impl<T: DeviceBase> Device for T {
    fn uuid(&self) -> Uuid {
        self.uuid_base()
    }

    fn list_methods(&self) -> Result<Vec<MethodInfo>, HlapiClientError> {
        let Response {
            id: _,
            generation: _,
            data: ResponseData::Methods { device_methods },
        } = self.client().make_request(Request {
            id: None,
            data: RequestData::Methods { uuid: self.uuid() },
        })?
        else {
            return Err(HlapiClientError::UnexpectedResponseType);
        };
        Ok(device_methods)
    }

    fn event_stream(&self) -> EventStream<T::Event> {
        EventStream::new(self.uuid(), self.client().clone(), Self::parse_event)
    }
}

#[derive(Debug)]
struct EventStreamInner<T> {
    events: Arc<Mutex<VecDeque<T>>>,
    client: HlapiClient,
    device_id: Uuid,
    listener_id: Uuid,
}

impl<T> Clone for EventStreamInner<T> {
    fn clone(&self) -> Self {
        Self {
            events: self.events.clone(),
            client: self.client.clone(),
            device_id: self.device_id,
            listener_id: self.listener_id,
        }
    }
}

impl<T: 'static + Send> EventStreamInner<T> {
    #[must_use]
    fn get_queued_event(&self) -> Option<T> {
        self.events.lock().unwrap().pop_front()
    }

    fn get_event(&self, timeout: Option<Duration>) -> Result<Option<T>, HlapiClientError> {
        let deadline = timeout.map(|t| Instant::now() + t);
        loop {
            if let Some(event) = self.get_queued_event() {
                return Ok(Some(event));
            }
            if !self
                .client
                .inner
                .lock()
                .map_err(|_| HlapiClientError::MutexPoisoned)?
                .process_event(deadline.map(|d| d.saturating_duration_since(Instant::now())))?
            {
                return Ok(self.get_queued_event());
            }
        }
    }

    fn clear(&self) {
        self.events.lock().unwrap().clear();
    }
}

impl<T> Drop for EventStreamInner<T> {
    fn drop(&mut self) {
        if let Ok(mut inner_client) = self.client.inner.lock() {
            inner_client.unregister_event_listener(self.device_id, self.listener_id);
        }
    }
}

/// A stream of events from a device
///
/// An [`EventStream`] keeps a queue of events received from its device. There is no bound to the
/// size of the queue, so streams should be fully processed or cleared regularly. Drop any streams
/// not in use so their queues do not build up forever.
#[derive(Debug)]
pub struct EventStream<T>(Arc<EventStreamInner<T>>);

impl<T> Clone for EventStream<T> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}

impl<T: 'static + Send> EventStream<T> {
    #[allow(clippy::type_complexity)]
    fn new(
        device_id: Uuid,
        client: HlapiClient,
        parse_fn: fn(&str, &JsonValue) -> Result<T, Box<dyn Error + Send + Sync>>,
    ) -> Self {
        let events = Arc::new(Mutex::new(VecDeque::new()));
        let listener_id = {
            let events = Arc::downgrade(&events);
            client.inner.lock().unwrap().register_event_listener(
                device_id,
                Box::new(move |typ, data| {
                    let event = (parse_fn)(typ, data)?;
                    if let Some(events) = events.upgrade() {
                        events.lock().unwrap().push_back(event);
                    }
                    Ok(())
                }),
            )
        };
        Self(Arc::new(EventStreamInner {
            events,
            client,
            device_id,
            listener_id,
        }))
    }

    /// Gets an event from the stream's queue if there is one
    #[must_use]
    pub fn get_queued_event(&self) -> Option<T> {
        self.0.get_queued_event()
    }

    /// Gets an event from the stream, optionally with a timeout. A zero duration means a non-blocking get,
    /// and [`None`] means wait forever. Returns [`None`] if the timeout was hit. The stream's client is
    /// unusable by other threads while this method is running.
    pub fn get_event(&self, timeout: Option<Duration>) -> Result<Option<T>, HlapiClientError> {
        self.0.get_event(timeout)
    }

    /// Clears the stream's event queue
    pub fn clear(&self) {
        self.0.clear();
    }
}

trait StreamHasEvent {
    fn has_queued_event(&self) -> bool;
}

impl<T> StreamHasEvent for EventStream<T> {
    fn has_queued_event(&self) -> bool {
        !self.0.events.lock().unwrap().is_empty()
    }
}

/// Allows waiting for events from any one of several streams on the same client
#[allow(missing_debug_implementations)]
pub struct EventStreamGroup {
    client: HlapiClient,
    streams: HashMap<(Uuid, Uuid), Box<dyn StreamHasEvent + Send + Sync>>,
}

impl EventStreamGroup {
    /// Creates a new poller for streams on the given client
    #[must_use]
    pub fn new(client: &HlapiClient) -> Self {
        Self {
            client: client.clone(),
            streams: HashMap::new(),
        }
    }

    /// Adds a stream to the list of streams to be polled. The stream *must* be on the same client as
    /// was passed to [`new`]. While a stream is part of a group, it is kept alive, so if a stream
    /// is not used anymore it should be [`remove`]d from all groups it is a part of.
    ///
    /// [`new`]: EventStreamGroup::new
    /// [`remove`]: EventStreamGroup::remove
    pub fn add<T: 'static + Send>(&mut self, stream: &EventStream<T>) {
        self.streams.insert(
            (stream.0.device_id, stream.0.listener_id),
            Box::new(stream.clone()),
        );
    }

    /// Removes a stream from the list of streams to be polled
    pub fn remove<T: 'static + Send>(&mut self, stream: &EventStream<T>) {
        self.streams
            .remove(&(stream.0.device_id, stream.0.listener_id));
    }

    /// Waits until one of the registered streams has an event, with an optional timeout. A zero
    /// duration means a non-blocking wait (i.e. just check if any stream has an event), and [`None`]
    /// means wait forever. Returns `Ok(true)` if one has an event, `Ok(false)` if the timeout was hit,
    /// and `Err` if there was an error. The stream's client is unusable by other threads while this
    /// method is running.
    pub fn wait_for_event(&self, timeout: Option<Duration>) -> Result<bool, HlapiClientError> {
        let deadline = timeout.map(|t| Instant::now() + t);
        loop {
            if self.streams.values().any(|x| x.has_queued_event()) {
                return Ok(true);
            }
            if !self
                .client
                .inner
                .lock()
                .map_err(|_| HlapiClientError::MutexPoisoned)?
                .process_event(deadline.map(|d| d.saturating_duration_since(Instant::now())))?
            {
                return Ok(false);
            }
        }
    }
}