rumqttc/

client.rs

//! This module offers a high level synchronous and asynchronous abstraction to
//! async eventloop.
use std::borrow::Cow;
use std::time::Duration;

use super::eventloop::{RequestChannelCapacity, RequestEnvelope};
use super::mqttbytes::QoS;
use super::mqttbytes::v5::{
    Auth, AuthProperties, AuthReasonCode, Filter, PubAck, PubRec, Publish, PublishProperties,
    Subscribe, SubscribeProperties, Unsubscribe, UnsubscribeProperties,
};
use super::{
    ConnectionError, Disconnect, DisconnectProperties, DisconnectReasonCode, Event, EventLoop,
    MqttOptions, Request,
};
use crate::notice::{AuthNoticeTx, PublishNoticeTx, SubscribeNoticeTx, UnsubscribeNoticeTx};
use crate::{
    AuthNotice, PublishNotice, SubscribeNotice, UnsubscribeNotice, valid_filter, valid_topic,
};

use bytes::Bytes;
use flume::{SendError, Sender, TrySendError};
use futures_util::FutureExt;
use tokio::runtime::{self, Runtime};
use tokio::time::timeout;

/// An error returned when a topic string fails validation against the MQTT specification.
#[derive(Debug, thiserror::Error, Clone, PartialEq, Eq)]
#[error("Invalid MQTT topic: '{0}'")]
pub struct InvalidTopic(String);

/// A newtype wrapper that guarantees its inner `String` is a valid MQTT topic.
///
/// This type prevents the cost of repeated validation for topics that are used
/// frequently. It can only be constructed via [`ValidatedTopic::new`], which
/// performs a one-time validation check.
///
/// Use this when publishing repeatedly to the same topic to avoid per-call
/// validation overhead in publish APIs that accept [`PublishTopic`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ValidatedTopic(String);

impl ValidatedTopic {
    /// Constructs a new `ValidatedTopic` after validating the input string.
    ///
    /// # Errors
    ///
    /// Returns [`InvalidTopic`] if the topic string does not conform to the MQTT specification.
    pub fn new<S: Into<String>>(topic: S) -> Result<Self, InvalidTopic> {
        let topic_string = topic.into();
        if valid_topic(&topic_string) {
            Ok(Self(topic_string))
        } else {
            Err(InvalidTopic(topic_string))
        }
    }
}

impl From<ValidatedTopic> for String {
    fn from(topic: ValidatedTopic) -> Self {
        topic.0
    }
}

/// Topic argument accepted by publish APIs.
///
/// `ValidatedTopic` variants skip per-call validation while string variants are
/// validated for MQTT topic correctness.
pub enum PublishTopic {
    /// Raw topic input that must be validated before publishing.
    Unvalidated(String),
    /// Topic that has already been validated once.
    Validated(ValidatedTopic),
}

impl PublishTopic {
    fn into_string_and_validation(self) -> (String, bool) {
        match self {
            Self::Unvalidated(topic) => (topic, true),
            Self::Validated(topic) => (topic.0, false),
        }
    }
}

impl From<ValidatedTopic> for PublishTopic {
    fn from(topic: ValidatedTopic) -> Self {
        Self::Validated(topic)
    }
}

impl From<String> for PublishTopic {
    fn from(topic: String) -> Self {
        Self::Unvalidated(topic)
    }
}

impl From<&str> for PublishTopic {
    fn from(topic: &str) -> Self {
        Self::Unvalidated(topic.to_owned())
    }
}

impl From<&String> for PublishTopic {
    fn from(topic: &String) -> Self {
        Self::Unvalidated(topic.clone())
    }
}

impl From<Cow<'_, str>> for PublishTopic {
    fn from(topic: Cow<'_, str>) -> Self {
        Self::Unvalidated(topic.into_owned())
    }
}

/// Client Error
#[derive(Debug, thiserror::Error)]
pub enum ClientError {
    #[error("Failed to send mqtt requests to eventloop")]
    Request(Box<Request>),
    #[error("Failed to send mqtt requests to eventloop")]
    TryRequest(Box<Request>),
    #[error("Tracked request API is unavailable for this client instance")]
    TrackingUnavailable,
}

impl From<SendError<Request>> for ClientError {
    fn from(e: SendError<Request>) -> Self {
        Self::Request(Box::new(e.into_inner()))
    }
}

impl From<TrySendError<Request>> for ClientError {
    fn from(e: TrySendError<Request>) -> Self {
        Self::TryRequest(Box::new(e.into_inner()))
    }
}

#[derive(Clone, Debug)]
enum RequestSender {
    Plain(Sender<Request>),
    WithNotice {
        requests: Sender<RequestEnvelope>,
        control_requests: Sender<RequestEnvelope>,
        immediate_disconnect: Sender<RequestEnvelope>,
    },
}

fn into_request(envelope: RequestEnvelope) -> Request {
    let (request, _notice) = envelope.into_parts();
    request
}

fn map_send_envelope_error(err: SendError<RequestEnvelope>) -> ClientError {
    ClientError::Request(Box::new(into_request(err.into_inner())))
}

fn map_try_send_envelope_error(err: TrySendError<RequestEnvelope>) -> ClientError {
    match err {
        TrySendError::Full(envelope) | TrySendError::Disconnected(envelope) => {
            ClientError::TryRequest(Box::new(into_request(envelope)))
        }
    }
}

const fn is_publish_request(request: &Request) -> bool {
    matches!(request, Request::Publish(_))
}

/// Prepared acknowledgement packet for manual acknowledgement mode.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ManualAck {
    PubAck(PubAck),
    PubRec(PubRec),
}

impl ManualAck {
    fn into_request(self) -> Request {
        match self {
            Self::PubAck(ack) => Request::PubAck(ack),
            Self::PubRec(rec) => Request::PubRec(rec),
        }
    }
}

/// An asynchronous client, communicates with MQTT `EventLoop`.
///
/// This is cloneable and can be used to asynchronously [`publish`](`AsyncClient::publish`),
/// [`subscribe`](`AsyncClient::subscribe`) through the `EventLoop`, which is to be polled parallelly.
///
/// **NOTE**: The `EventLoop` must be regularly polled in order to send, receive and process packets
/// from the broker, i.e. move ahead.
///
/// Bounded clients apply backpressure through the client request channel. If the
/// same task that drives [`EventLoop::poll`](crate::EventLoop::poll) awaits
/// request-sending APIs such as [`publish`](Self::publish),
/// [`subscribe`](Self::subscribe), [`unsubscribe`](Self::unsubscribe), or
/// [`ack`](Self::ack) while that channel is full, it can self-block: the send is
/// waiting for the event loop to read a request, but the event loop cannot make
/// progress until that same task polls it again. For bounded async clients,
/// prefer driving the event loop in a dedicated task. Use [`try_publish`](Self::try_publish)
/// when dropping outgoing publishes under overload is intended.
///
/// The request channel is an admission queue, not a strict global wire FIFO
/// guarantee. Under publish flow-control pressure, non-`PUBLISH` control
/// packets can pass earlier `QoS` 1/ `QoS` 2 publishes that are not currently
/// sendable. Application publishes preserve FIFO with other publishes.
#[derive(Clone, Debug)]
pub struct AsyncClient {
    request_tx: RequestSender,
}

/// Builder for synchronous MQTT clients.
///
/// The request channel is bounded by default using
/// [`MqttOptions::request_channel_capacity`]. Use [`Self::capacity`] to override
/// the bounded capacity, or [`Self::unbounded`] to opt into an unbounded request channel.
#[derive(Debug)]
pub struct ClientBuilder {
    options: MqttOptions,
    capacity: RequestChannelCapacity,
}

/// Builder for asynchronous MQTT clients.
///
/// The request channel is bounded by default using
/// [`MqttOptions::request_channel_capacity`]. Use [`Self::capacity`] to override
/// the bounded capacity, or [`Self::unbounded`] to opt into an unbounded request channel.
#[derive(Debug)]
pub struct AsyncClientBuilder {
    options: MqttOptions,
    capacity: RequestChannelCapacity,
}

#[must_use]
fn build_async_client(
    options: MqttOptions,
    capacity: RequestChannelCapacity,
) -> (AsyncClient, EventLoop) {
    let (eventloop, request_tx, control_request_tx, immediate_disconnect_tx) =
        EventLoop::new_for_async_client_with_capacity(options, capacity);
    let client = AsyncClient {
        request_tx: RequestSender::WithNotice {
            requests: request_tx,
            control_requests: control_request_tx,
            immediate_disconnect: immediate_disconnect_tx,
        },
    };

    (client, eventloop)
}

impl ClientBuilder {
    /// Create a new builder for a synchronous [`Client`].
    #[must_use]
    pub const fn new(options: MqttOptions) -> Self {
        let capacity = RequestChannelCapacity::Bounded(options.request_channel_capacity());
        Self { options, capacity }
    }

    /// Use a bounded request channel with the given capacity.
    ///
    /// `0` creates a bounded zero-capacity rendezvous channel. Use [`Self::unbounded`]
    /// for an unbounded request channel.
    #[must_use]
    pub const fn capacity(mut self, cap: usize) -> Self {
        self.capacity = RequestChannelCapacity::Bounded(cap);
        self
    }

    /// Use an unbounded request channel.
    #[must_use]
    pub const fn unbounded(mut self) -> Self {
        self.capacity = RequestChannelCapacity::Unbounded;
        self
    }

    /// Build a synchronous client and connection.
    ///
    /// This builder always produces the synchronous client pair so the
    /// terminal `build()` method matches the entry point that created it.
    ///
    /// # Panics
    ///
    /// Panics if the current-thread Tokio runtime cannot be created.
    #[must_use]
    pub fn build(self) -> (Client, Connection) {
        let (client, eventloop) = build_async_client(self.options, self.capacity);
        let client = Client { client };

        let runtime = runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        let connection = Connection::new(eventloop, runtime);
        (client, connection)
    }
}

impl AsyncClientBuilder {
    /// Create a new builder for an asynchronous [`AsyncClient`].
    #[must_use]
    pub const fn new(options: MqttOptions) -> Self {
        let capacity = RequestChannelCapacity::Bounded(options.request_channel_capacity());
        Self { options, capacity }
    }

    /// Use a bounded request channel with the given capacity.
    ///
    /// `0` creates a bounded zero-capacity rendezvous channel. Use [`Self::unbounded`]
    /// for an unbounded request channel.
    #[must_use]
    pub const fn capacity(mut self, cap: usize) -> Self {
        self.capacity = RequestChannelCapacity::Bounded(cap);
        self
    }

    /// Use an unbounded request channel.
    #[must_use]
    pub const fn unbounded(mut self) -> Self {
        self.capacity = RequestChannelCapacity::Unbounded;
        self
    }

    /// Build an asynchronous client and event loop.
    ///
    /// This builder always produces the asynchronous client pair so the
    /// terminal `build()` method matches the entry point that created it.
    #[must_use]
    pub fn build(self) -> (AsyncClient, EventLoop) {
        build_async_client(self.options, self.capacity)
    }
}

impl AsyncClient {
    /// Create a builder for an [`AsyncClient`].
    ///
    /// The returned [`AsyncClientBuilder`] only builds the asynchronous
    /// client pair, which keeps the terminal `build()` method aligned with
    /// this entry point.
    #[must_use]
    pub const fn builder(options: MqttOptions) -> AsyncClientBuilder {
        AsyncClientBuilder::new(options)
    }

    /// Create a new `AsyncClient` from a channel `Sender`.
    ///
    /// This is mostly useful for creating a test instance where you can
    /// listen on the corresponding receiver.
    #[must_use]
    pub const fn from_senders(request_tx: Sender<Request>) -> Self {
        Self {
            request_tx: RequestSender::Plain(request_tx),
        }
    }

    async fn send_request_async(&self, request: Request) -> Result<(), ClientError> {
        match &self.request_tx {
            RequestSender::Plain(tx) => tx.send_async(request).await.map_err(ClientError::from),
            RequestSender::WithNotice {
                requests,
                control_requests,
                ..
            } => {
                let tx = if is_publish_request(&request) {
                    requests
                } else {
                    control_requests
                };
                tx.send_async(RequestEnvelope::plain(request))
                    .await
                    .map_err(map_send_envelope_error)
            }
        }
    }

    fn try_send_request(&self, request: Request) -> Result<(), ClientError> {
        match &self.request_tx {
            RequestSender::Plain(tx) => tx.try_send(request).map_err(ClientError::from),
            RequestSender::WithNotice {
                requests,
                control_requests,
                ..
            } => {
                let tx = if is_publish_request(&request) {
                    requests
                } else {
                    control_requests
                };
                tx.try_send(RequestEnvelope::plain(request))
                    .map_err(map_try_send_envelope_error)
            }
        }
    }

    fn send_request(&self, request: Request) -> Result<(), ClientError> {
        match &self.request_tx {
            RequestSender::Plain(tx) => tx.send(request).map_err(ClientError::from),
            RequestSender::WithNotice {
                requests,
                control_requests,
                ..
            } => {
                let tx = if is_publish_request(&request) {
                    requests
                } else {
                    control_requests
                };
                tx.send(RequestEnvelope::plain(request))
                    .map_err(map_send_envelope_error)
            }
        }
    }

    async fn send_immediate_disconnect_async(&self, request: Request) -> Result<(), ClientError> {
        match &self.request_tx {
            RequestSender::Plain(tx) => tx.send_async(request).await.map_err(ClientError::from),
            RequestSender::WithNotice {
                immediate_disconnect,
                ..
            } => immediate_disconnect
                .send_async(RequestEnvelope::plain(request))
                .await
                .map_err(map_send_envelope_error),
        }
    }

    fn send_immediate_disconnect(&self, request: Request) -> Result<(), ClientError> {
        match &self.request_tx {
            RequestSender::Plain(tx) => tx.send(request).map_err(ClientError::from),
            RequestSender::WithNotice {
                immediate_disconnect,
                ..
            } => immediate_disconnect
                .send(RequestEnvelope::plain(request))
                .map_err(map_send_envelope_error),
        }
    }

    fn try_send_immediate_disconnect(&self, request: Request) -> Result<(), ClientError> {
        match &self.request_tx {
            RequestSender::Plain(tx) => tx.try_send(request).map_err(ClientError::from),
            RequestSender::WithNotice {
                immediate_disconnect,
                ..
            } => immediate_disconnect
                .try_send(RequestEnvelope::plain(request))
                .map_err(map_try_send_envelope_error),
        }
    }

    async fn send_tracked_publish_async(
        &self,
        publish: Publish,
    ) -> Result<PublishNotice, ClientError> {
        let RequestSender::WithNotice {
            requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = PublishNoticeTx::new();
        request_tx
            .send_async(RequestEnvelope::tracked_publish(publish, notice_tx))
            .await
            .map_err(map_send_envelope_error)?;
        Ok(notice)
    }

    fn try_send_tracked_publish(&self, publish: Publish) -> Result<PublishNotice, ClientError> {
        let RequestSender::WithNotice {
            requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = PublishNoticeTx::new();
        request_tx
            .try_send(RequestEnvelope::tracked_publish(publish, notice_tx))
            .map_err(map_try_send_envelope_error)?;
        Ok(notice)
    }

    async fn send_tracked_subscribe_async(
        &self,
        subscribe: Subscribe,
    ) -> Result<SubscribeNotice, ClientError> {
        let RequestSender::WithNotice {
            control_requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = SubscribeNoticeTx::new();
        request_tx
            .send_async(RequestEnvelope::tracked_subscribe(subscribe, notice_tx))
            .await
            .map_err(map_send_envelope_error)?;
        Ok(notice)
    }

    fn try_send_tracked_subscribe(
        &self,
        subscribe: Subscribe,
    ) -> Result<SubscribeNotice, ClientError> {
        let RequestSender::WithNotice {
            control_requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = SubscribeNoticeTx::new();
        request_tx
            .try_send(RequestEnvelope::tracked_subscribe(subscribe, notice_tx))
            .map_err(map_try_send_envelope_error)?;
        Ok(notice)
    }

    async fn send_tracked_unsubscribe_async(
        &self,
        unsubscribe: Unsubscribe,
    ) -> Result<UnsubscribeNotice, ClientError> {
        let RequestSender::WithNotice {
            control_requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = UnsubscribeNoticeTx::new();
        request_tx
            .send_async(RequestEnvelope::tracked_unsubscribe(unsubscribe, notice_tx))
            .await
            .map_err(map_send_envelope_error)?;
        Ok(notice)
    }

    fn try_send_tracked_unsubscribe(
        &self,
        unsubscribe: Unsubscribe,
    ) -> Result<UnsubscribeNotice, ClientError> {
        let RequestSender::WithNotice {
            control_requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = UnsubscribeNoticeTx::new();
        request_tx
            .try_send(RequestEnvelope::tracked_unsubscribe(unsubscribe, notice_tx))
            .map_err(map_try_send_envelope_error)?;
        Ok(notice)
    }

    async fn send_tracked_auth_async(&self, auth: Auth) -> Result<AuthNotice, ClientError> {
        let RequestSender::WithNotice {
            control_requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = AuthNoticeTx::new();
        request_tx
            .send_async(RequestEnvelope::tracked_auth(auth, notice_tx))
            .await
            .map_err(map_send_envelope_error)?;
        Ok(notice)
    }

    fn send_tracked_auth(&self, auth: Auth) -> Result<AuthNotice, ClientError> {
        let RequestSender::WithNotice {
            control_requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = AuthNoticeTx::new();
        request_tx
            .send(RequestEnvelope::tracked_auth(auth, notice_tx))
            .map_err(map_send_envelope_error)?;
        Ok(notice)
    }

    fn try_send_tracked_auth(&self, auth: Auth) -> Result<AuthNotice, ClientError> {
        let RequestSender::WithNotice {
            control_requests: request_tx,
            ..
        } = &self.request_tx
        else {
            return Err(ClientError::TrackingUnavailable);
        };

        let (notice_tx, notice) = AuthNoticeTx::new();
        request_tx
            .try_send(RequestEnvelope::tracked_auth(auth, notice_tx))
            .map_err(map_try_send_envelope_error)?;
        Ok(notice)
    }

    /// Sends a MQTT Publish to the `EventLoop`.
    async fn handle_publish<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: Option<PublishProperties>,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        let (topic, needs_validation) = topic.into().into_string_and_validation();
        let invalid_topic = (needs_validation && !valid_topic(&topic))
            || empty_topic_without_valid_alias(&topic, properties.as_ref());
        let mut publish = Publish::new(topic, qos, payload, properties);
        publish.retain = retain;
        let publish = Request::Publish(publish);

        if invalid_topic {
            return Err(ClientError::Request(Box::new(publish)));
        }

        self.send_request_async(publish).await?;
        Ok(())
    }

    async fn handle_publish_tracked<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: Option<PublishProperties>,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        let (topic, needs_validation) = topic.into().into_string_and_validation();
        let invalid_topic = (needs_validation && !valid_topic(&topic))
            || empty_topic_without_valid_alias(&topic, properties.as_ref());
        let mut publish = Publish::new(topic, qos, payload, properties);
        publish.retain = retain;
        let request = Request::Publish(publish.clone());

        if invalid_topic {
            return Err(ClientError::Request(Box::new(request)));
        }

        self.send_tracked_publish_async(publish).await
    }

    /// Sends a MQTT Publish with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish_with_properties<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: PublishProperties,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_publish(topic, qos, retain, payload, Some(properties))
            .await
    }

    /// Sends a MQTT Publish with properties to the `EventLoop` and returns a tracked notice.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish_with_properties_tracked<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: PublishProperties,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_publish_tracked(topic, qos, retain, payload, Some(properties))
            .await
    }

    /// Sends a MQTT Publish to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_publish(topic, qos, retain, payload, None).await
    }

    /// Sends a MQTT Publish to the `EventLoop` and returns a tracked notice.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish_tracked<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_publish_tracked(topic, qos, retain, payload, None)
            .await
    }

    /// Attempts to send a MQTT Publish to the `EventLoop`.
    fn handle_try_publish<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: Option<PublishProperties>,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        let (topic, needs_validation) = topic.into().into_string_and_validation();
        let invalid_topic = (needs_validation && !valid_topic(&topic))
            || empty_topic_without_valid_alias(&topic, properties.as_ref());
        let mut publish = Publish::new(topic, qos, payload, properties);
        publish.retain = retain;
        let publish = Request::Publish(publish);

        if invalid_topic {
            return Err(ClientError::TryRequest(Box::new(publish)));
        }

        self.try_send_request(publish)?;
        Ok(())
    }

    fn handle_try_publish_tracked<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: Option<PublishProperties>,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        let (topic, needs_validation) = topic.into().into_string_and_validation();
        let invalid_topic = (needs_validation && !valid_topic(&topic))
            || empty_topic_without_valid_alias(&topic, properties.as_ref());
        let mut publish = Publish::new(topic, qos, payload, properties);
        publish.retain = retain;
        let request = Request::Publish(publish.clone());

        if invalid_topic {
            return Err(ClientError::TryRequest(Box::new(request)));
        }

        self.try_send_tracked_publish(publish)
    }

    /// Attempts to send a MQTT Publish with properties to the `EventLoop`.
    ///
    /// This is the non-blocking publish API for overload policies that may drop
    /// outgoing publishes. If the bounded request channel is full, this returns
    /// an error immediately and the publish has not been queued.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued immediately on the event loop.
    pub fn try_publish_with_properties<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: PublishProperties,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_try_publish(topic, qos, retain, payload, Some(properties))
    }

    /// Attempts to send a MQTT Publish with properties to the `EventLoop` and returns a tracked notice.
    ///
    /// This is the non-blocking tracked publish API for overload policies that
    /// may drop outgoing publishes. If the bounded request channel is full, this
    /// returns an error immediately and the publish has not been queued.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued immediately on the event loop.
    pub fn try_publish_with_properties_tracked<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: PublishProperties,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_try_publish_tracked(topic, qos, retain, payload, Some(properties))
    }

    /// Attempts to send a MQTT Publish to the `EventLoop`.
    ///
    /// This is the non-blocking publish API for overload policies that may drop
    /// outgoing publishes. If the bounded request channel is full, this returns
    /// an error immediately and the publish has not been queued.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued immediately on the event loop.
    pub fn try_publish<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_try_publish(topic, qos, retain, payload, None)
    }

    /// Attempts to send a MQTT Publish to the `EventLoop` and returns a tracked notice.
    ///
    /// This is the non-blocking tracked publish API for overload policies that
    /// may drop outgoing publishes. If the bounded request channel is full, this
    /// returns an error immediately and the publish has not been queued.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued immediately on the event loop.
    pub fn try_publish_tracked<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_try_publish_tracked(topic, qos, retain, payload, None)
    }

    /// Prepares a MQTT PubAck/PubRec packet for manual acknowledgement.
    ///
    /// Returns `None` for `QoS0` publishes, which do not require acknowledgement.
    ///
    /// This is typically used together with
    /// [`MqttOptions::set_manual_acks`](`crate::MqttOptions::set_manual_acks`)
    /// when acknowledgement must be deferred or customized.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use rumqttc::mqttbytes::v5::{PubAckProperties, PubAckReason, Publish};
    /// use rumqttc::{AsyncClient, ManualAck, MqttOptions};
    ///
    /// # async fn example(client: AsyncClient, publish: Publish) -> Result<(), rumqttc::ClientError> {
    /// let Some(mut ack) = client.prepare_ack(&publish) else {
    ///     return Ok(());
    /// };
    ///
    /// if let ManualAck::PubAck(puback) = &mut ack {
    ///     puback.reason = PubAckReason::NoMatchingSubscribers;
    ///     puback.properties = Some(PubAckProperties {
    ///         reason_string: Some("No active subscribers now".to_owned()),
    ///         user_properties: vec![("source".to_owned(), "application".to_owned())],
    ///     });
    /// }
    ///
    /// client.manual_ack(ack).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub const fn prepare_ack(&self, publish: &Publish) -> Option<ManualAck> {
        prepare_ack(publish)
    }

    /// Sends a prepared MQTT PubAck/PubRec to the `EventLoop`.
    ///
    /// This is useful when `manual_acks` is enabled and acknowledgement must be deferred.
    ///
    /// # Errors
    ///
    /// Returns an error if the acknowledgement cannot be queued on the event
    /// loop.
    pub async fn manual_ack(&self, ack: ManualAck) -> Result<(), ClientError> {
        self.send_request_async(ack.into_request()).await?;
        Ok(())
    }

    /// Attempts to send a prepared MQTT PubAck/PubRec to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the acknowledgement cannot be queued immediately on
    /// the event loop.
    pub fn try_manual_ack(&self, ack: ManualAck) -> Result<(), ClientError> {
        self.try_send_request(ack.into_request())?;
        Ok(())
    }

    /// Sends a MQTT PubAck/PubRec to the `EventLoop` based on publish `QoS`.
    /// Only needed if the `manual_acks` flag is set.
    ///
    /// # Errors
    ///
    /// Returns an error if the derived acknowledgement cannot be queued on the
    /// event loop.
    pub async fn ack(&self, publish: &Publish) -> Result<(), ClientError> {
        if let Some(ack) = self.prepare_ack(publish) {
            self.manual_ack(ack).await?;
        }
        Ok(())
    }

    /// Attempts to send a MQTT PubAck/PubRec to the `EventLoop` based on publish `QoS`.
    /// Only needed if the `manual_acks` flag is set.
    ///
    /// # Errors
    ///
    /// Returns an error if the derived acknowledgement cannot be queued
    /// immediately on the event loop.
    pub fn try_ack(&self, publish: &Publish) -> Result<(), ClientError> {
        if let Some(ack) = self.prepare_ack(publish) {
            self.try_manual_ack(ack)?;
        }
        Ok(())
    }

    /// Sends a MQTT AUTH packet to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued on the event loop.
    pub async fn reauth(&self, properties: Option<AuthProperties>) -> Result<(), ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        let auth = Request::Auth(auth);
        self.send_request_async(auth).await?;
        Ok(())
    }

    /// Sends a tracked MQTT re-authentication request to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued on the event loop
    /// or if tracked request notices are unavailable for this client instance.
    pub async fn reauth_tracked(
        &self,
        properties: Option<AuthProperties>,
    ) -> Result<AuthNotice, ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        self.send_tracked_auth_async(auth).await
    }

    /// Attempts to send a MQTT AUTH packet to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued immediately on the
    /// event loop.
    pub fn try_reauth(&self, properties: Option<AuthProperties>) -> Result<(), ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        let auth = Request::Auth(auth);
        self.try_send_request(auth)?;
        Ok(())
    }

    /// Attempts to send a tracked MQTT re-authentication request to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued immediately on the
    /// event loop or if tracked request notices are unavailable for this client
    /// instance.
    pub fn try_reauth_tracked(
        &self,
        properties: Option<AuthProperties>,
    ) -> Result<AuthNotice, ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        self.try_send_tracked_auth(auth)
    }

    /// Sends a MQTT Publish to the `EventLoop`
    async fn handle_publish_bytes<T>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: Bytes,
        properties: Option<PublishProperties>,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
    {
        let (topic, needs_validation) = topic.into().into_string_and_validation();
        let invalid_topic = (needs_validation && !valid_topic(&topic))
            || empty_topic_without_valid_alias(&topic, properties.as_ref());
        let mut publish = Publish::new(topic, qos, payload, properties);
        publish.retain = retain;
        let publish = Request::Publish(publish);

        if invalid_topic {
            return Err(ClientError::Request(Box::new(publish)));
        }

        self.send_request_async(publish).await?;
        Ok(())
    }

    async fn handle_publish_bytes_tracked<T>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: Bytes,
        properties: Option<PublishProperties>,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
    {
        let (topic, needs_validation) = topic.into().into_string_and_validation();
        let invalid_topic = (needs_validation && !valid_topic(&topic))
            || empty_topic_without_valid_alias(&topic, properties.as_ref());
        let mut publish = Publish::new(topic, qos, payload, properties);
        publish.retain = retain;
        let request = Request::Publish(publish.clone());

        if invalid_topic {
            return Err(ClientError::Request(Box::new(request)));
        }

        self.send_tracked_publish_async(publish).await
    }

    /// Sends a MQTT Publish with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish_bytes_with_properties<T>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: Bytes,
        properties: PublishProperties,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
    {
        self.handle_publish_bytes(topic, qos, retain, payload, Some(properties))
            .await
    }

    /// Sends a MQTT Publish with `Bytes` payload and properties, returning a tracked notice.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish_bytes_with_properties_tracked<T>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: Bytes,
        properties: PublishProperties,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
    {
        self.handle_publish_bytes_tracked(topic, qos, retain, payload, Some(properties))
            .await
    }

    /// Sends a MQTT Publish with `Bytes` payload to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish_bytes<T>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: Bytes,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
    {
        self.handle_publish_bytes(topic, qos, retain, payload, None)
            .await
    }

    /// Sends a MQTT Publish with `Bytes` payload to the `EventLoop` and returns a tracked notice.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub async fn publish_bytes_tracked<T>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: Bytes,
    ) -> Result<PublishNotice, ClientError>
    where
        T: Into<PublishTopic>,
    {
        self.handle_publish_bytes_tracked(topic, qos, retain, payload, None)
            .await
    }

    /// Sends a MQTT Subscribe to the `EventLoop`
    async fn handle_subscribe<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: Option<SubscribeProperties>,
    ) -> Result<(), ClientError> {
        let filter = Filter::new(topic, qos);
        let subscribe = Subscribe::new(filter, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::Request(Box::new(subscribe.into())));
        }

        self.send_request_async(subscribe.into()).await?;
        Ok(())
    }

    async fn handle_subscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: Option<SubscribeProperties>,
    ) -> Result<SubscribeNotice, ClientError> {
        let filter = Filter::new(topic, qos);
        let subscribe = Subscribe::new(filter, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::Request(Box::new(subscribe.into())));
        }

        self.send_tracked_subscribe_async(subscribe).await
    }

    /// Sends a MQTT Subscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued on the event loop.
    pub async fn subscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError> {
        self.handle_subscribe(topic, qos, Some(properties)).await
    }

    /// Sends a tracked MQTT Subscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued on the event loop.
    pub async fn subscribe_with_properties_tracked<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: SubscribeProperties,
    ) -> Result<SubscribeNotice, ClientError> {
        self.handle_subscribe_tracked(topic, qos, Some(properties))
            .await
    }

    /// Sends a MQTT Subscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued on the event loop.
    pub async fn subscribe<S: Into<String>>(&self, topic: S, qos: QoS) -> Result<(), ClientError> {
        self.handle_subscribe(topic, qos, None).await
    }

    /// Sends a tracked MQTT Subscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued on the event loop.
    pub async fn subscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
    ) -> Result<SubscribeNotice, ClientError> {
        self.handle_subscribe_tracked(topic, qos, None).await
    }

    /// Attempts to send a MQTT Subscribe to the `EventLoop`
    fn handle_try_subscribe<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: Option<SubscribeProperties>,
    ) -> Result<(), ClientError> {
        let filter = Filter::new(topic, qos);
        let subscribe = Subscribe::new(filter, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::TryRequest(Box::new(subscribe.into())));
        }

        self.try_send_request(subscribe.into())?;
        Ok(())
    }

    fn handle_try_subscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: Option<SubscribeProperties>,
    ) -> Result<SubscribeNotice, ClientError> {
        let filter = Filter::new(topic, qos);
        let subscribe = Subscribe::new(filter, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::TryRequest(Box::new(subscribe.into())));
        }

        self.try_send_tracked_subscribe(subscribe)
    }

    /// Attempts to send a MQTT Subscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued immediately on the event loop.
    pub fn try_subscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError> {
        self.handle_try_subscribe(topic, qos, Some(properties))
    }

    /// Attempts to send a tracked MQTT Subscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued immediately on the event loop.
    pub fn try_subscribe_with_properties_tracked<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: SubscribeProperties,
    ) -> Result<SubscribeNotice, ClientError> {
        self.handle_try_subscribe_tracked(topic, qos, Some(properties))
    }

    /// Attempts to send a MQTT Subscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued immediately on the event loop.
    pub fn try_subscribe<S: Into<String>>(&self, topic: S, qos: QoS) -> Result<(), ClientError> {
        self.handle_try_subscribe(topic, qos, None)
    }

    /// Attempts to send a tracked MQTT Subscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued immediately on the event loop.
    pub fn try_subscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
    ) -> Result<SubscribeNotice, ClientError> {
        self.handle_try_subscribe_tracked(topic, qos, None)
    }

    /// Sends a MQTT Subscribe for multiple topics to the `EventLoop`
    async fn handle_subscribe_many<T>(
        &self,
        topics: T,
        properties: Option<SubscribeProperties>,
    ) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        let subscribe = Subscribe::new_many(topics, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::Request(Box::new(subscribe.into())));
        }

        self.send_request_async(subscribe.into()).await?;

        Ok(())
    }

    async fn handle_subscribe_many_tracked<T>(
        &self,
        topics: T,
        properties: Option<SubscribeProperties>,
    ) -> Result<SubscribeNotice, ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        let subscribe = Subscribe::new_many(topics, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::Request(Box::new(subscribe.into())));
        }

        self.send_tracked_subscribe_async(subscribe).await
    }

    /// Sends a MQTT Subscribe for multiple topics with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued on the event loop.
    pub async fn subscribe_many_with_properties<T>(
        &self,
        topics: T,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_subscribe_many(topics, Some(properties)).await
    }

    /// Sends a tracked MQTT Subscribe for multiple topics with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued on the event loop.
    pub async fn subscribe_many_with_properties_tracked<T>(
        &self,
        topics: T,
        properties: SubscribeProperties,
    ) -> Result<SubscribeNotice, ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_subscribe_many_tracked(topics, Some(properties))
            .await
    }

    /// Sends a MQTT Subscribe for multiple topics to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued on the event loop.
    pub async fn subscribe_many<T>(&self, topics: T) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_subscribe_many(topics, None).await
    }

    /// Sends a tracked MQTT Subscribe for multiple topics to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued on the event loop.
    pub async fn subscribe_many_tracked<T>(&self, topics: T) -> Result<SubscribeNotice, ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_subscribe_many_tracked(topics, None).await
    }

    /// Attempts to send a MQTT Subscribe for multiple topics to the `EventLoop`
    fn handle_try_subscribe_many<T>(
        &self,
        topics: T,
        properties: Option<SubscribeProperties>,
    ) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        let subscribe = Subscribe::new_many(topics, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::TryRequest(Box::new(subscribe.into())));
        }

        self.try_send_request(subscribe.into())?;
        Ok(())
    }

    fn handle_try_subscribe_many_tracked<T>(
        &self,
        topics: T,
        properties: Option<SubscribeProperties>,
    ) -> Result<SubscribeNotice, ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        let subscribe = Subscribe::new_many(topics, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::TryRequest(Box::new(subscribe.into())));
        }

        self.try_send_tracked_subscribe(subscribe)
    }

    /// Attempts to send a MQTT Subscribe for multiple topics with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued immediately on the event loop.
    pub fn try_subscribe_many_with_properties<T>(
        &self,
        topics: T,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_try_subscribe_many(topics, Some(properties))
    }

    /// Attempts to send a tracked MQTT Subscribe for multiple topics with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued immediately on the event loop.
    pub fn try_subscribe_many_with_properties_tracked<T>(
        &self,
        topics: T,
        properties: SubscribeProperties,
    ) -> Result<SubscribeNotice, ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_try_subscribe_many_tracked(topics, Some(properties))
    }

    /// Attempts to send a MQTT Subscribe for multiple topics to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued immediately on the event loop.
    pub fn try_subscribe_many<T>(&self, topics: T) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_try_subscribe_many(topics, None)
    }

    /// Attempts to send a tracked MQTT Subscribe for multiple topics to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued immediately on the event loop.
    pub fn try_subscribe_many_tracked<T>(&self, topics: T) -> Result<SubscribeNotice, ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_try_subscribe_many_tracked(topics, None)
    }

    /// Sends a MQTT Unsubscribe to the `EventLoop`
    async fn handle_unsubscribe<S: Into<String>>(
        &self,
        topic: S,
        properties: Option<UnsubscribeProperties>,
    ) -> Result<(), ClientError> {
        let unsubscribe = Unsubscribe::new(topic, properties);
        let request = Request::Unsubscribe(unsubscribe);
        self.send_request_async(request).await?;
        Ok(())
    }

    async fn handle_unsubscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
        properties: Option<UnsubscribeProperties>,
    ) -> Result<UnsubscribeNotice, ClientError> {
        let unsubscribe = Unsubscribe::new(topic, properties);
        self.send_tracked_unsubscribe_async(unsubscribe).await
    }

    /// Sends a MQTT Unsubscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued on the event loop.
    pub async fn unsubscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        properties: UnsubscribeProperties,
    ) -> Result<(), ClientError> {
        self.handle_unsubscribe(topic, Some(properties)).await
    }

    /// Sends a tracked MQTT Unsubscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued on the event loop.
    pub async fn unsubscribe_with_properties_tracked<S: Into<String>>(
        &self,
        topic: S,
        properties: UnsubscribeProperties,
    ) -> Result<UnsubscribeNotice, ClientError> {
        self.handle_unsubscribe_tracked(topic, Some(properties))
            .await
    }

    /// Sends a MQTT Unsubscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued on the event loop.
    pub async fn unsubscribe<S: Into<String>>(&self, topic: S) -> Result<(), ClientError> {
        self.handle_unsubscribe(topic, None).await
    }

    /// Sends a tracked MQTT Unsubscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued on the event loop.
    pub async fn unsubscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
    ) -> Result<UnsubscribeNotice, ClientError> {
        self.handle_unsubscribe_tracked(topic, None).await
    }

    /// Attempts to send a MQTT Unsubscribe to the `EventLoop`
    fn handle_try_unsubscribe<S: Into<String>>(
        &self,
        topic: S,
        properties: Option<UnsubscribeProperties>,
    ) -> Result<(), ClientError> {
        let unsubscribe = Unsubscribe::new(topic, properties);
        let request = Request::Unsubscribe(unsubscribe);
        self.try_send_request(request)?;
        Ok(())
    }

    fn handle_try_unsubscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
        properties: Option<UnsubscribeProperties>,
    ) -> Result<UnsubscribeNotice, ClientError> {
        let unsubscribe = Unsubscribe::new(topic, properties);
        self.try_send_tracked_unsubscribe(unsubscribe)
    }

    /// Attempts to send a MQTT Unsubscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued immediately on the
    /// event loop.
    pub fn try_unsubscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        properties: UnsubscribeProperties,
    ) -> Result<(), ClientError> {
        self.handle_try_unsubscribe(topic, Some(properties))
    }

    /// Attempts to send a tracked MQTT Unsubscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued immediately on the
    /// event loop.
    pub fn try_unsubscribe_with_properties_tracked<S: Into<String>>(
        &self,
        topic: S,
        properties: UnsubscribeProperties,
    ) -> Result<UnsubscribeNotice, ClientError> {
        self.handle_try_unsubscribe_tracked(topic, Some(properties))
    }

    /// Attempts to send a MQTT Unsubscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued immediately on the
    /// event loop.
    pub fn try_unsubscribe<S: Into<String>>(&self, topic: S) -> Result<(), ClientError> {
        self.handle_try_unsubscribe(topic, None)
    }

    /// Attempts to send a tracked MQTT Unsubscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued immediately on the
    /// event loop.
    pub fn try_unsubscribe_tracked<S: Into<String>>(
        &self,
        topic: S,
    ) -> Result<UnsubscribeNotice, ClientError> {
        self.handle_try_unsubscribe_tracked(topic, None)
    }

    /// Queues a graceful MQTT disconnect barrier with default
    /// `DisconnectReasonCode::NormalDisconnection`.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub async fn disconnect(&self) -> Result<(), ClientError> {
        self.handle_disconnect(DisconnectReasonCode::NormalDisconnection, None)
            .await
    }

    /// Queues a graceful MQTT disconnect barrier with properties.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub async fn disconnect_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.handle_disconnect(reason, Some(properties)).await
    }

    /// Queues a graceful MQTT disconnect barrier with a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub async fn disconnect_with_timeout(&self, timeout: Duration) -> Result<(), ClientError> {
        self.handle_disconnect_with_timeout(
            DisconnectReasonCode::NormalDisconnection,
            None,
            timeout,
        )
        .await
    }

    /// Queues a graceful MQTT disconnect barrier with properties and a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub async fn disconnect_with_properties_timeout(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
        timeout: Duration,
    ) -> Result<(), ClientError> {
        self.handle_disconnect_with_timeout(reason, Some(properties), timeout)
            .await
    }

    /// Sends a MQTT disconnect immediately without waiting for in-flight requests.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub async fn disconnect_now(&self) -> Result<(), ClientError> {
        self.handle_disconnect_now(DisconnectReasonCode::NormalDisconnection, None)
            .await
    }

    /// Sends a MQTT disconnect with properties immediately without waiting for in-flight requests.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub async fn disconnect_now_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.handle_disconnect_now(reason, Some(properties)).await
    }

    // Handle disconnect interface which can have properties or not
    async fn handle_disconnect(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Result<(), ClientError> {
        let request = Self::build_disconnect_request(reason, properties);
        self.send_request_async(request).await?;
        Ok(())
    }

    async fn handle_disconnect_with_timeout(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
        timeout: Duration,
    ) -> Result<(), ClientError> {
        let disconnect = Self::build_disconnect_packet(reason, properties);
        self.send_request_async(Request::DisconnectWithTimeout(disconnect, timeout))
            .await?;
        Ok(())
    }

    async fn handle_disconnect_now(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Result<(), ClientError> {
        let disconnect = Self::build_disconnect_packet(reason, properties);
        self.send_immediate_disconnect_async(Request::DisconnectNow(disconnect))
            .await?;
        Ok(())
    }

    /// Attempts to queue a graceful MQTT disconnect barrier with default
    /// `DisconnectReasonCode::NormalDisconnection`.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect(&self) -> Result<(), ClientError> {
        self.handle_try_disconnect(DisconnectReasonCode::NormalDisconnection, None)
    }

    /// Attempts to queue a graceful MQTT disconnect barrier with properties.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.handle_try_disconnect(reason, Some(properties))
    }

    /// Attempts to queue a graceful MQTT disconnect with a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_with_timeout(&self, timeout: Duration) -> Result<(), ClientError> {
        self.handle_try_disconnect_with_timeout(
            DisconnectReasonCode::NormalDisconnection,
            None,
            timeout,
        )
    }

    /// Attempts to queue a graceful MQTT disconnect with properties and a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_with_properties_timeout(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
        timeout: Duration,
    ) -> Result<(), ClientError> {
        self.handle_try_disconnect_with_timeout(reason, Some(properties), timeout)
    }

    /// Attempts to queue an immediate MQTT disconnect.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_now(&self) -> Result<(), ClientError> {
        self.handle_try_disconnect_now(DisconnectReasonCode::NormalDisconnection, None)
    }

    /// Attempts to queue an immediate MQTT disconnect with properties.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_now_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.handle_try_disconnect_now(reason, Some(properties))
    }

    // Handle disconnect interface which can have properties or not
    fn handle_try_disconnect(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Result<(), ClientError> {
        let request = Self::build_disconnect_request(reason, properties);
        self.try_send_request(request)?;
        Ok(())
    }

    fn handle_try_disconnect_with_timeout(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
        timeout: Duration,
    ) -> Result<(), ClientError> {
        let disconnect = Self::build_disconnect_packet(reason, properties);
        self.try_send_request(Request::DisconnectWithTimeout(disconnect, timeout))?;
        Ok(())
    }

    fn handle_try_disconnect_now(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Result<(), ClientError> {
        let disconnect = Self::build_disconnect_packet(reason, properties);
        self.try_send_immediate_disconnect(Request::DisconnectNow(disconnect))?;
        Ok(())
    }

    // Helper function to build disconnect request
    fn build_disconnect_request(
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Request {
        Request::Disconnect(Self::build_disconnect_packet(reason, properties))
    }

    fn build_disconnect_packet(
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Disconnect {
        properties.map_or_else(
            || Disconnect::new(reason),
            |p| Disconnect::new_with_properties(reason, p),
        )
    }
}

const fn prepare_ack(publish: &Publish) -> Option<ManualAck> {
    let ack = match publish.qos {
        QoS::AtMostOnce => return None,
        QoS::AtLeastOnce => ManualAck::PubAck(PubAck::new(publish.pkid, None)),
        QoS::ExactlyOnce => ManualAck::PubRec(PubRec::new(publish.pkid, None)),
    };
    Some(ack)
}

/// A synchronous client, communicates with MQTT `EventLoop`.
///
/// This is cloneable and can be used to synchronously [`publish`](`AsyncClient::publish`),
/// [`subscribe`](`AsyncClient::subscribe`) through the `EventLoop`/`Connection`, which is to be polled in parallel
/// by iterating over the object returned by [`Connection.iter()`](Connection::iter) in a separate thread.
///
/// **NOTE**: The `EventLoop`/`Connection` must be regularly polled(`.next()` in case of `Connection`) in order
/// to send, receive and process packets from the broker, i.e. move ahead.
///
/// An asynchronous channel handle can also be extracted if necessary.
#[derive(Clone)]
pub struct Client {
    client: AsyncClient,
}

impl Client {
    /// Create a builder for a [`Client`].
    ///
    /// The returned [`ClientBuilder`] only builds the synchronous client
    /// pair, which keeps the terminal `build()` method aligned with this
    /// entry point.
    #[must_use]
    pub const fn builder(options: MqttOptions) -> ClientBuilder {
        ClientBuilder::new(options)
    }

    /// Create a new `Client` from a channel `Sender`.
    ///
    /// This is mostly useful for creating a test instance where you can
    /// listen on the corresponding receiver.
    #[must_use]
    pub const fn from_sender(request_tx: Sender<Request>) -> Self {
        Self {
            client: AsyncClient::from_senders(request_tx),
        }
    }

    /// Sends a MQTT Publish to the `EventLoop`
    fn handle_publish<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: Option<PublishProperties>,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        let (topic, needs_validation) = topic.into().into_string_and_validation();
        let invalid_topic = (needs_validation && !valid_topic(&topic))
            || empty_topic_without_valid_alias(&topic, properties.as_ref());
        let mut publish = Publish::new(topic, qos, payload, properties);
        publish.retain = retain;
        let request = Request::Publish(publish);

        if invalid_topic {
            return Err(ClientError::Request(Box::new(request)));
        }

        self.client.send_request(request)?;
        Ok(())
    }

    /// Sends a MQTT Publish with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub fn publish_with_properties<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: PublishProperties,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_publish(topic, qos, retain, payload, Some(properties))
    }

    /// Sends a MQTT Publish to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued on the event loop.
    pub fn publish<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.handle_publish(topic, qos, retain, payload, None)
    }

    /// Attempts to send a MQTT Publish with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued immediately on the event loop.
    pub fn try_publish_with_properties<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
        properties: PublishProperties,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.client
            .try_publish_with_properties(topic, qos, retain, payload, properties)
    }

    /// Attempts to send a MQTT Publish to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic or topic alias usage is invalid, or if
    /// the request cannot be queued immediately on the event loop.
    pub fn try_publish<T, P>(
        &self,
        topic: T,
        qos: QoS,
        retain: bool,
        payload: P,
    ) -> Result<(), ClientError>
    where
        T: Into<PublishTopic>,
        P: Into<Bytes>,
    {
        self.client.try_publish(topic, qos, retain, payload)
    }

    /// Prepares a MQTT PubAck/PubRec packet for manual acknowledgement.
    ///
    /// Returns `None` for `QoS0` publishes, which do not require acknowledgement.
    pub const fn prepare_ack(&self, publish: &Publish) -> Option<ManualAck> {
        self.client.prepare_ack(publish)
    }

    /// Sends a prepared MQTT PubAck/PubRec to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the acknowledgement cannot be queued on the event
    /// loop.
    pub fn manual_ack(&self, ack: ManualAck) -> Result<(), ClientError> {
        self.client.send_request(ack.into_request())?;
        Ok(())
    }

    /// Attempts to send a prepared MQTT PubAck/PubRec to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the acknowledgement cannot be queued immediately on
    /// the event loop.
    pub fn try_manual_ack(&self, ack: ManualAck) -> Result<(), ClientError> {
        self.client.try_manual_ack(ack)?;
        Ok(())
    }

    /// Sends a MQTT PubAck/PubRec to the `EventLoop` based on publish `QoS`.
    /// Only needed if the `manual_acks` flag is set.
    ///
    /// # Errors
    ///
    /// Returns an error if the derived acknowledgement cannot be queued on the
    /// event loop.
    pub fn ack(&self, publish: &Publish) -> Result<(), ClientError> {
        if let Some(ack) = self.prepare_ack(publish) {
            self.manual_ack(ack)?;
        }
        Ok(())
    }

    /// Attempts to send a MQTT PubAck/PubRec to the `EventLoop` based on publish `QoS`.
    /// Only needed if the `manual_acks` flag is set.
    ///
    /// # Errors
    ///
    /// Returns an error if the derived acknowledgement cannot be queued
    /// immediately on the event loop.
    pub fn try_ack(&self, publish: &Publish) -> Result<(), ClientError> {
        if let Some(ack) = self.prepare_ack(publish) {
            self.try_manual_ack(ack)?;
        }
        Ok(())
    }

    /// Sends a MQTT AUTH packet to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued on the event loop.
    pub fn reauth(&self, properties: Option<AuthProperties>) -> Result<(), ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        let auth = Request::Auth(auth);
        self.client.send_request(auth)?;
        Ok(())
    }

    /// Sends a tracked MQTT re-authentication request to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued on the event loop
    /// or if tracked request notices are unavailable for this client instance.
    pub fn reauth_tracked(
        &self,
        properties: Option<AuthProperties>,
    ) -> Result<AuthNotice, ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        self.client.send_tracked_auth(auth)
    }

    /// Attempts to send a MQTT AUTH packet to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued immediately on the
    /// event loop.
    pub fn try_reauth(&self, properties: Option<AuthProperties>) -> Result<(), ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        let auth = Request::Auth(auth);
        self.client.try_send_request(auth)?;
        Ok(())
    }

    /// Attempts to send a tracked MQTT re-authentication request to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the AUTH packet cannot be queued immediately on the
    /// event loop or if tracked request notices are unavailable for this client
    /// instance.
    pub fn try_reauth_tracked(
        &self,
        properties: Option<AuthProperties>,
    ) -> Result<AuthNotice, ClientError> {
        let auth = Auth::new(AuthReasonCode::ReAuthenticate, properties);
        self.client.try_send_tracked_auth(auth)
    }

    /// Sends a MQTT Subscribe to the `EventLoop`
    fn handle_subscribe<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: Option<SubscribeProperties>,
    ) -> Result<(), ClientError> {
        let filter = Filter::new(topic, qos);
        let subscribe = Subscribe::new(filter, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::Request(Box::new(subscribe.into())));
        }

        self.client.send_request(subscribe.into())?;
        Ok(())
    }

    /// Sends a MQTT Subscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued on the event loop.
    pub fn subscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError> {
        self.handle_subscribe(topic, qos, Some(properties))
    }

    /// Sends a MQTT Subscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued on the event loop.
    pub fn subscribe<S: Into<String>>(&self, topic: S, qos: QoS) -> Result<(), ClientError> {
        self.handle_subscribe(topic, qos, None)
    }

    /// Attempts to send a MQTT Subscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued immediately on the event loop.
    pub fn try_subscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        qos: QoS,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError> {
        self.client
            .try_subscribe_with_properties(topic, qos, properties)
    }

    /// Attempts to send a MQTT Subscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the topic filter is invalid or if the request
    /// cannot be queued immediately on the event loop.
    pub fn try_subscribe<S: Into<String>>(&self, topic: S, qos: QoS) -> Result<(), ClientError> {
        self.client.try_subscribe(topic, qos)
    }

    /// Sends a MQTT Subscribe for multiple topics to the `EventLoop`
    fn handle_subscribe_many<T>(
        &self,
        topics: T,
        properties: Option<SubscribeProperties>,
    ) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        let subscribe = Subscribe::new_many(topics, properties);
        if !subscribe_has_valid_filters(&subscribe) {
            return Err(ClientError::Request(Box::new(subscribe.into())));
        }

        self.client.send_request(subscribe.into())?;
        Ok(())
    }

    /// Sends a MQTT Subscribe for multiple topics with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued on the event loop.
    pub fn subscribe_many_with_properties<T>(
        &self,
        topics: T,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_subscribe_many(topics, Some(properties))
    }

    /// Sends a MQTT Subscribe for multiple topics to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued on the event loop.
    pub fn subscribe_many<T>(&self, topics: T) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.handle_subscribe_many(topics, None)
    }

    /// Attempts to send a MQTT Subscribe for multiple topics with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued immediately on the event loop.
    pub fn try_subscribe_many_with_properties<T>(
        &self,
        topics: T,
        properties: SubscribeProperties,
    ) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.client
            .try_subscribe_many_with_properties(topics, properties)
    }

    /// Attempts to send a MQTT Subscribe for multiple topics to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the filter list is invalid or if the request cannot
    /// be queued immediately on the event loop.
    pub fn try_subscribe_many<T>(&self, topics: T) -> Result<(), ClientError>
    where
        T: IntoIterator<Item = Filter>,
    {
        self.client.try_subscribe_many(topics)
    }

    /// Sends a MQTT Unsubscribe to the `EventLoop`
    fn handle_unsubscribe<S: Into<String>>(
        &self,
        topic: S,
        properties: Option<UnsubscribeProperties>,
    ) -> Result<(), ClientError> {
        let unsubscribe = Unsubscribe::new(topic, properties);
        let request = Request::Unsubscribe(unsubscribe);
        self.client.send_request(request)?;
        Ok(())
    }

    /// Sends a MQTT Unsubscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued on the event loop.
    pub fn unsubscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        properties: UnsubscribeProperties,
    ) -> Result<(), ClientError> {
        self.handle_unsubscribe(topic, Some(properties))
    }

    /// Sends a MQTT Unsubscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued on the event loop.
    pub fn unsubscribe<S: Into<String>>(&self, topic: S) -> Result<(), ClientError> {
        self.handle_unsubscribe(topic, None)
    }

    /// Attempts to send a MQTT Unsubscribe with properties to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued immediately on the
    /// event loop.
    pub fn try_unsubscribe_with_properties<S: Into<String>>(
        &self,
        topic: S,
        properties: UnsubscribeProperties,
    ) -> Result<(), ClientError> {
        self.client
            .try_unsubscribe_with_properties(topic, properties)
    }

    /// Attempts to send a MQTT Unsubscribe to the `EventLoop`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request cannot be queued immediately on the
    /// event loop.
    pub fn try_unsubscribe<S: Into<String>>(&self, topic: S) -> Result<(), ClientError> {
        self.client.try_unsubscribe(topic)
    }

    /// Queues a graceful MQTT disconnect barrier.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub fn disconnect(&self) -> Result<(), ClientError> {
        self.handle_disconnect(DisconnectReasonCode::NormalDisconnection, None)
    }

    /// Queues a graceful MQTT disconnect barrier with properties.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub fn disconnect_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.handle_disconnect(reason, Some(properties))
    }

    /// Queues a graceful MQTT disconnect barrier with a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub fn disconnect_with_timeout(&self, timeout: Duration) -> Result<(), ClientError> {
        self.handle_disconnect_with_timeout(
            DisconnectReasonCode::NormalDisconnection,
            None,
            timeout,
        )
    }

    /// Queues a graceful MQTT disconnect barrier with properties and a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub fn disconnect_with_properties_timeout(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
        timeout: Duration,
    ) -> Result<(), ClientError> {
        self.handle_disconnect_with_timeout(reason, Some(properties), timeout)
    }

    /// Sends a MQTT disconnect immediately without waiting for in-flight requests.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub fn disconnect_now(&self) -> Result<(), ClientError> {
        self.handle_disconnect_now(DisconnectReasonCode::NormalDisconnection, None)
    }

    /// Sends a MQTT disconnect with properties immediately without waiting for in-flight requests.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued on the
    /// event loop.
    pub fn disconnect_now_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.handle_disconnect_now(reason, Some(properties))
    }

    fn handle_disconnect(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Result<(), ClientError> {
        let request = AsyncClient::build_disconnect_request(reason, properties);
        self.client.send_request(request)?;
        Ok(())
    }

    fn handle_disconnect_with_timeout(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
        timeout: Duration,
    ) -> Result<(), ClientError> {
        let disconnect = AsyncClient::build_disconnect_packet(reason, properties);
        self.client
            .send_request(Request::DisconnectWithTimeout(disconnect, timeout))?;
        Ok(())
    }

    fn handle_disconnect_now(
        &self,
        reason: DisconnectReasonCode,
        properties: Option<DisconnectProperties>,
    ) -> Result<(), ClientError> {
        let disconnect = AsyncClient::build_disconnect_packet(reason, properties);
        self.client
            .send_immediate_disconnect(Request::DisconnectNow(disconnect))?;
        Ok(())
    }

    /// Attempts to queue a graceful MQTT disconnect barrier.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect(&self) -> Result<(), ClientError> {
        self.client.try_disconnect()
    }

    /// Attempts to queue a graceful MQTT disconnect barrier with properties.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// for previously accepted outbound `QoS` 1/ `QoS` 2 publishes and tracked
    /// subscribe/unsubscribe requests to complete before sending MQTT
    /// `DISCONNECT`.
    ///
    /// This request uses the normal client request channel. Under publish
    /// flow-control pressure, it may pass earlier `QoS` 1/ `QoS` 2 publishes
    /// that are not currently sendable; once observed, it becomes the graceful
    /// drain barrier.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.client.handle_try_disconnect(reason, Some(properties))
    }

    /// Attempts to queue a graceful MQTT disconnect with a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_with_timeout(&self, timeout: Duration) -> Result<(), ClientError> {
        self.client.try_disconnect_with_timeout(timeout)
    }

    /// Attempts to queue a graceful MQTT disconnect with properties and a drain timeout.
    ///
    /// Once the event loop observes this request, it stops processing later
    /// application work, flushes previously accepted `QoS` 0 publishes, and waits
    /// up to `timeout` for previously accepted outbound `QoS` 1/ `QoS` 2 publishes
    /// and tracked subscribe/unsubscribe requests to complete. `QoS` 1 publishes
    /// complete on `PUBACK`, `QoS` 2 publishes complete on `PUBCOMP`, tracked
    /// subscribes complete on `SUBACK`, and tracked unsubscribes complete on
    /// `UNSUBACK`.
    ///
    /// If the drain completes before the deadline, the event loop sends and
    /// flushes MQTT `DISCONNECT`. If the deadline expires first, polling returns
    /// `ConnectionError::DisconnectTimeout` and MQTT `DISCONNECT` is not sent.
    ///
    /// This request uses the normal client request channel. The timeout starts
    /// only after the event loop observes this request, not necessarily when
    /// this method queues it.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_with_properties_timeout(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
        timeout: Duration,
    ) -> Result<(), ClientError> {
        self.client
            .handle_try_disconnect_with_timeout(reason, Some(properties), timeout)
    }

    /// Attempts to queue an immediate MQTT disconnect.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_now(&self) -> Result<(), ClientError> {
        self.client.try_disconnect_now()
    }

    /// Attempts to queue an immediate MQTT disconnect with properties.
    ///
    /// This request uses a dedicated immediate shutdown channel, not the normal
    /// application request channel. It may bypass queued application work and
    /// does not wait for unresolved `QoS` 1/ `QoS` 2 publish handshakes.
    ///
    /// # Errors
    ///
    /// Returns an error if the disconnect request cannot be queued
    /// immediately on the event loop.
    pub fn try_disconnect_now_with_properties(
        &self,
        reason: DisconnectReasonCode,
        properties: DisconnectProperties,
    ) -> Result<(), ClientError> {
        self.client
            .handle_try_disconnect_now(reason, Some(properties))
    }
}

#[must_use]
fn empty_topic_without_valid_alias(topic: &str, properties: Option<&PublishProperties>) -> bool {
    topic.is_empty()
        && properties
            .and_then(|props| props.topic_alias)
            .unwrap_or_default()
            == 0
}

#[must_use]
fn subscribe_has_valid_filters(subscribe: &Subscribe) -> bool {
    !subscribe.filters.is_empty()
        && subscribe
            .filters
            .iter()
            .all(|filter| valid_filter(&filter.path))
}

/// Error type returned by [`Connection::recv`]
#[derive(Debug, Eq, PartialEq)]
pub struct RecvError;

/// Error type returned by [`Connection::try_recv`]
#[derive(Debug, Eq, PartialEq)]
pub enum TryRecvError {
    /// User has closed requests channel
    Disconnected,
    /// Did not resolve
    Empty,
}

/// Error type returned by [`Connection::recv_timeout`]
#[derive(Debug, Eq, PartialEq)]
pub enum RecvTimeoutError {
    /// User has closed requests channel
    Disconnected,
    /// Recv request timedout
    Timeout,
}

///  MQTT connection. Maintains all the necessary state
pub struct Connection {
    pub eventloop: EventLoop,
    runtime: Runtime,
}
impl Connection {
    const fn new(eventloop: EventLoop, runtime: Runtime) -> Self {
        Self { eventloop, runtime }
    }

    /// Returns an iterator over this connection. Iterating over this is all that's
    /// necessary to make connection progress and maintain a robust connection.
    /// Just continuing to loop will reconnect
    /// **NOTE** Don't block this while iterating
    // ideally this should be named iter_mut because it requires a mutable reference
    // Also we can implement IntoIter for this to make it easy to iterate over it
    #[must_use = "Connection should be iterated over a loop to make progress"]
    pub const fn iter(&mut self) -> Iter<'_> {
        Iter { connection: self }
    }

    /// Attempt to fetch an incoming [`Event`] on the [`EventLoop`], returning an error
    /// if all clients/users have closed requests channel.
    ///
    /// [`EventLoop`]: super::EventLoop
    ///
    /// # Errors
    ///
    /// Returns [`RecvError`] if all request senders have been dropped.
    pub fn recv(&mut self) -> Result<Result<Event, ConnectionError>, RecvError> {
        let f = self.eventloop.poll();
        let event = self.runtime.block_on(f);

        resolve_event(event).ok_or(RecvError)
    }

    /// Attempt to fetch an incoming [`Event`] on the [`EventLoop`], returning an error
    /// if none immediately present or all clients/users have closed requests channel.
    ///
    /// [`EventLoop`]: super::EventLoop
    ///
    /// # Errors
    ///
    /// Returns [`TryRecvError::Empty`] if no event is immediately ready, or
    /// [`TryRecvError::Disconnected`] if all request senders have been dropped.
    pub fn try_recv(&mut self) -> Result<Result<Event, ConnectionError>, TryRecvError> {
        let f = self.eventloop.poll();
        // Enters the runtime context so we can poll the future, as required by `now_or_never()`.
        // ref: https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#method.enter
        let _guard = self.runtime.enter();
        let event = f.now_or_never().ok_or(TryRecvError::Empty)?;

        resolve_event(event).ok_or(TryRecvError::Disconnected)
    }

    /// Attempt to fetch an incoming [`Event`] on the [`EventLoop`], returning an error
    /// if all clients/users have closed requests channel or the timeout has expired.
    ///
    /// [`EventLoop`]: super::EventLoop
    ///
    /// # Errors
    ///
    /// Returns [`RecvTimeoutError::Timeout`] if no event arrives before
    /// `duration`, or [`RecvTimeoutError::Disconnected`] if all request
    /// senders have been dropped.
    pub fn recv_timeout(
        &mut self,
        duration: Duration,
    ) -> Result<Result<Event, ConnectionError>, RecvTimeoutError> {
        let f = self.eventloop.poll();
        let event = self
            .runtime
            .block_on(async { timeout(duration, f).await })
            .map_err(|_| RecvTimeoutError::Timeout)?;

        resolve_event(event).ok_or(RecvTimeoutError::Disconnected)
    }
}

fn resolve_event(event: Result<Event, ConnectionError>) -> Option<Result<Event, ConnectionError>> {
    match event {
        Ok(v) => Some(Ok(v)),
        // closing of request channel should stop the iterator
        Err(ConnectionError::RequestsDone) => {
            trace!("Done with requests");
            None
        }
        Err(e) => Some(Err(e)),
    }
}

/// Iterator which polls the `EventLoop` for connection progress
pub struct Iter<'a> {
    connection: &'a mut Connection,
}

impl Iterator for Iter<'_> {
    type Item = Result<Event, ConnectionError>;

    fn next(&mut self) -> Option<Self::Item> {
        self.connection.recv().ok()
    }
}

#[cfg(test)]
mod test {
    use crate::mqttbytes::v5::{
        LastWill, PubAckProperties, PubAckReason, PubRecProperties, PubRecReason,
    };

    use super::*;

    #[test]
    fn calling_iter_twice_on_connection_shouldnt_panic() {
        let mut mqttoptions = MqttOptions::new("test-1", "localhost");
        let will = LastWill::new("hello/world", "good bye", QoS::AtMostOnce, false, None);
        mqttoptions.set_keep_alive(5).set_last_will(will);

        let (_, mut connection) = Client::builder(mqttoptions).capacity(10).build();
        let _ = connection.iter();
        let _ = connection.iter();
    }

    #[test]
    fn builder_uses_options_request_channel_capacity_by_default() {
        let mut mqttoptions = MqttOptions::new("test-1", "localhost");
        mqttoptions.set_request_channel_capacity(1);
        let builder: AsyncClientBuilder = AsyncClient::builder(mqttoptions);
        let (client, _eventloop) = builder.build();

        client
            .try_publish("hello/world", QoS::AtMostOnce, false, "one")
            .expect("first request should fit configured capacity");
        assert!(matches!(
            client.try_publish("hello/world", QoS::AtMostOnce, false, "two"),
            Err(ClientError::TryRequest(request)) if matches!(*request, Request::Publish(_))
        ));
    }

    #[test]
    fn sync_and_async_entry_points_return_distinct_builder_types() {
        let sync_builder = Client::builder(MqttOptions::new("test-sync", "localhost"));
        let async_builder = AsyncClient::builder(MqttOptions::new("test-async", "localhost"));

        let _: ClientBuilder = sync_builder;
        let _: AsyncClientBuilder = async_builder;
    }

    #[test]
    fn builder_capacity_overrides_options_request_channel_capacity() {
        let mut mqttoptions = MqttOptions::new("test-1", "localhost");
        mqttoptions.set_request_channel_capacity(1);
        let (client, _eventloop) = Client::builder(mqttoptions).capacity(2).build();

        client
            .try_publish("hello/world", QoS::AtMostOnce, false, "one")
            .expect("first request should fit overridden capacity");
        client
            .try_publish("hello/world", QoS::AtMostOnce, false, "two")
            .expect("second request should fit overridden capacity");
        assert!(matches!(
            client.try_publish("hello/world", QoS::AtMostOnce, false, "three"),
            Err(ClientError::TryRequest(request)) if matches!(*request, Request::Publish(_))
        ));
    }

    #[test]
    fn builder_capacity_zero_is_bounded_rendezvous() {
        let mqttoptions = MqttOptions::new("test-1", "localhost");
        let (client, _eventloop) = AsyncClient::builder(mqttoptions).capacity(0).build();

        assert!(matches!(
            client.try_publish("hello/world", QoS::AtMostOnce, false, "one"),
            Err(ClientError::TryRequest(request)) if matches!(*request, Request::Publish(_))
        ));
    }

    #[test]
    fn unbounded_builder_allows_try_publish_without_polling() {
        let mqttoptions = MqttOptions::new("test-1", "localhost");
        let (client, _eventloop) = AsyncClient::builder(mqttoptions).unbounded().build();

        for i in 0..128 {
            client
                .try_publish("hello/world", QoS::AtMostOnce, false, vec![i])
                .expect("unbounded channel should accept requests without polling");
        }
    }

    #[tokio::test]
    async fn bounded_publish_blocks_when_channel_is_full_without_polling() {
        let mqttoptions = MqttOptions::new("test-1", "localhost");
        let (client, _eventloop) = AsyncClient::builder(mqttoptions).capacity(1).build();

        client
            .publish("hello/world", QoS::AtMostOnce, false, "one")
            .await
            .expect("first request should fit bounded channel");

        let result = tokio::time::timeout(
            std::time::Duration::from_millis(25),
            client.publish("hello/world", QoS::AtMostOnce, false, "two"),
        )
        .await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn unbounded_publish_completes_without_polling() {
        let mqttoptions = MqttOptions::new("test-1", "localhost");
        let (client, _eventloop) = AsyncClient::builder(mqttoptions).unbounded().build();

        for i in 0..128 {
            client
                .publish("hello/world", QoS::AtMostOnce, false, vec![i])
                .await
                .expect("unbounded channel should accept requests without polling");
        }
    }

    #[test]
    fn should_be_able_to_build_test_client_from_channel() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);
        client
            .publish("hello/world", QoS::ExactlyOnce, false, "good bye")
            .expect("Should be able to publish");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn prepare_ack_maps_qos_to_manual_ack_packets_v5() {
        let (tx, _) = flume::bounded(1);
        let client = Client::from_sender(tx);

        let qos0 = Publish::new("hello/world", QoS::AtMostOnce, vec![1], None);
        assert!(client.prepare_ack(&qos0).is_none());

        let mut qos1 = Publish::new("hello/world", QoS::AtLeastOnce, vec![1], None);
        qos1.pkid = 7;
        match client.prepare_ack(&qos1) {
            Some(ManualAck::PubAck(ack)) => assert_eq!(ack.pkid, 7),
            ack => panic!("expected QoS1 PubAck, got {ack:?}"),
        }

        let mut qos2 = Publish::new("hello/world", QoS::ExactlyOnce, vec![1], None);
        qos2.pkid = 9;
        match client.prepare_ack(&qos2) {
            Some(ManualAck::PubRec(ack)) => assert_eq!(ack.pkid, 9),
            ack => panic!("expected QoS2 PubRec, got {ack:?}"),
        }
    }

    #[test]
    fn manual_ack_sends_custom_puback_reason_and_properties() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);

        let expected_properties = PubAckProperties {
            reason_string: Some("no downstream subscribers".to_owned()),
            user_properties: vec![("source".to_owned(), "unit-test".to_owned())],
        };
        let mut ack = PubAck::new(41, None);
        ack.reason = PubAckReason::NoMatchingSubscribers;
        ack.properties = Some(expected_properties.clone());

        client
            .manual_ack(ManualAck::PubAck(ack))
            .expect("manual_ack should send request");

        let request = rx.try_recv().expect("Should have ack request");
        match request {
            Request::PubAck(ack) => {
                assert_eq!(ack.pkid, 41);
                assert_eq!(ack.reason, PubAckReason::NoMatchingSubscribers);
                assert_eq!(ack.properties, Some(expected_properties));
            }
            request => panic!("Expected PubAck request, got {request:?}"),
        }
    }

    #[test]
    fn try_manual_ack_sends_custom_pubrec_reason_and_properties() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);

        let expected_properties = PubRecProperties {
            reason_string: Some("queued for qos2 flow".to_owned()),
            user_properties: vec![("source".to_owned(), "unit-test".to_owned())],
        };
        let mut ack = PubRec::new(52, None);
        ack.reason = PubRecReason::ImplementationSpecificError;
        ack.properties = Some(expected_properties.clone());

        client
            .try_manual_ack(ManualAck::PubRec(ack))
            .expect("try_manual_ack should send request");

        let request = rx.try_recv().expect("Should have ack request");
        match request {
            Request::PubRec(ack) => {
                assert_eq!(ack.pkid, 52);
                assert_eq!(ack.reason, PubRecReason::ImplementationSpecificError);
                assert_eq!(ack.properties, Some(expected_properties));
            }
            request => panic!("Expected PubRec request, got {request:?}"),
        }
    }

    #[test]
    fn ack_and_try_ack_send_default_success_packets_v5() {
        let (tx, rx) = flume::bounded(2);
        let client = Client::from_sender(tx);

        let mut qos1 = Publish::new("hello/world", QoS::AtLeastOnce, vec![1], None);
        qos1.pkid = 11;
        client.ack(&qos1).expect("ack should send PubAck");

        let mut qos2 = Publish::new("hello/world", QoS::ExactlyOnce, vec![1], None);
        qos2.pkid = 13;
        client
            .try_ack(&qos2)
            .expect("try_ack should send PubRec request");

        let first = rx.try_recv().expect("Should receive first ack request");
        match first {
            Request::PubAck(ack) => {
                assert_eq!(ack.pkid, 11);
                assert_eq!(ack.reason, PubAckReason::Success);
                assert_eq!(ack.properties, None);
            }
            request => panic!("Expected PubAck request, got {request:?}"),
        }

        let second = rx.try_recv().expect("Should receive second ack request");
        match second {
            Request::PubRec(ack) => {
                assert_eq!(ack.pkid, 13);
                assert_eq!(ack.reason, PubRecReason::Success);
                assert_eq!(ack.properties, None);
            }
            request => panic!("Expected PubRec request, got {request:?}"),
        }
    }

    #[test]
    fn test_reauth() {
        let (client, mut connection) = Client::builder(MqttOptions::new("test-1", "localhost"))
            .capacity(10)
            .build();
        let props = AuthProperties {
            method: Some("test".to_string()),
            data: Some(Bytes::from("test")),
            reason: None,
            user_properties: vec![],
        };
        client
            .reauth(Some(props.clone()))
            .expect("Should be able to reauth");
        let _ = connection.iter().next().expect("Should have event");

        client
            .try_reauth(Some(props))
            .expect("Should be able to reauth");
        let _ = connection.iter().next().expect("Should have event");
    }

    #[test]
    fn can_publish_with_validated_topic() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let valid_topic = ValidatedTopic::new("hello/world").unwrap();
        client
            .publish(valid_topic, QoS::ExactlyOnce, false, "good bye")
            .expect("Should be able to publish");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn publish_accepts_borrowed_string_topic() {
        let (tx, rx) = flume::bounded(2);
        let client = Client::from_sender(tx);
        let topic = "hello/world".to_string();
        client
            .publish(&topic, QoS::ExactlyOnce, false, "good bye")
            .expect("Should be able to publish");
        client
            .try_publish(&topic, QoS::ExactlyOnce, false, "good bye")
            .expect("Should be able to publish");
        let _ = rx.try_recv().expect("Should have message");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn publish_accepts_cow_topic_variants() {
        let (tx, rx) = flume::bounded(2);
        let client = Client::from_sender(tx);
        client
            .publish(
                std::borrow::Cow::Borrowed("hello/world"),
                QoS::ExactlyOnce,
                false,
                "good bye",
            )
            .expect("Should be able to publish");
        client
            .try_publish(
                std::borrow::Cow::Owned("hello/world".to_owned()),
                QoS::ExactlyOnce,
                false,
                "good bye",
            )
            .expect("Should be able to publish");
        let _ = rx.try_recv().expect("Should have message");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn publishing_invalid_cow_topic_fails() {
        let (tx, _) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let err = client
            .publish(
                std::borrow::Cow::Borrowed("a/+/b"),
                QoS::ExactlyOnce,
                false,
                "good bye",
            )
            .expect_err("Invalid publish topic should fail");
        assert!(matches!(err, ClientError::Request(req) if matches!(*req, Request::Publish(_))));
    }

    #[test]
    fn validated_topic_ergonomics() {
        let valid_topic = ValidatedTopic::new("hello/world").unwrap();
        let valid_topic_can_be_cloned = valid_topic.clone();
        // ValidatedTopic can be compared
        assert_eq!(valid_topic, valid_topic_can_be_cloned);
    }

    #[test]
    fn creating_invalid_validated_topic_fails() {
        assert_eq!(
            ValidatedTopic::new("a/+/b"),
            Err(InvalidTopic("a/+/b".to_string()))
        );
    }

    #[test]
    fn publish_with_properties_accepts_validated_topic() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let valid_topic = ValidatedTopic::new("hello/world").unwrap();
        client
            .publish_with_properties(
                valid_topic,
                QoS::ExactlyOnce,
                false,
                "good bye",
                PublishProperties::default(),
            )
            .expect("Should be able to publish");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn publish_with_properties_empty_topic_requires_nonzero_alias() {
        let (tx, _) = flume::bounded(1);
        let client = Client::from_sender(tx);

        let err = client
            .publish_with_properties(
                "",
                QoS::AtMostOnce,
                false,
                "good bye",
                PublishProperties::default(),
            )
            .expect_err("Empty topic without topic alias should fail");
        assert!(matches!(err, ClientError::Request(req) if matches!(*req, Request::Publish(_))));

        let err = client
            .publish_with_properties(
                "",
                QoS::AtMostOnce,
                false,
                "good bye",
                PublishProperties {
                    topic_alias: Some(0),
                    ..Default::default()
                },
            )
            .expect_err("Empty topic with topic alias 0 should fail");
        assert!(matches!(err, ClientError::Request(req) if matches!(*req, Request::Publish(_))));
    }

    #[test]
    fn publish_with_properties_empty_topic_accepts_nonzero_alias() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);

        client
            .publish_with_properties(
                "",
                QoS::AtMostOnce,
                false,
                "good bye",
                PublishProperties {
                    topic_alias: Some(1),
                    ..Default::default()
                },
            )
            .expect("Empty topic with non-zero topic alias should be accepted");

        let request = rx.try_recv().expect("Should have message");
        match request {
            Request::Publish(publish) => {
                assert!(publish.topic.is_empty());
                assert_eq!(
                    publish
                        .properties
                        .as_ref()
                        .and_then(|properties| properties.topic_alias),
                    Some(1)
                );
            }
            request => panic!("Expected Publish request, got {request:?}"),
        }
    }

    #[test]
    fn try_publish_accepts_validated_topic() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let valid_topic = ValidatedTopic::new("hello/world").unwrap();
        client
            .try_publish(valid_topic, QoS::ExactlyOnce, false, "good bye")
            .expect("Should be able to publish");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn try_publish_with_properties_accepts_validated_topic() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let valid_topic = ValidatedTopic::new("hello/world").unwrap();
        client
            .try_publish_with_properties(
                valid_topic,
                QoS::ExactlyOnce,
                false,
                "good bye",
                PublishProperties::default(),
            )
            .expect("Should be able to publish");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn try_publish_with_properties_empty_topic_requires_nonzero_alias() {
        let (tx, _) = flume::bounded(1);
        let client = Client::from_sender(tx);

        let err = client
            .try_publish_with_properties(
                "",
                QoS::AtMostOnce,
                false,
                "good bye",
                PublishProperties::default(),
            )
            .expect_err("Empty topic without topic alias should fail");
        assert!(matches!(err, ClientError::TryRequest(req) if matches!(*req, Request::Publish(_))));
    }

    #[test]
    fn publishing_invalid_raw_topic_fails() {
        let (tx, _) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let err = client
            .publish("a/+/b", QoS::ExactlyOnce, false, "good bye")
            .expect_err("Invalid publish topic should fail");
        assert!(matches!(err, ClientError::Request(req) if matches!(*req, Request::Publish(_))));
    }

    #[test]
    fn async_publish_paths_accept_validated_topic() {
        let (tx, rx) = flume::bounded(4);
        let client = AsyncClient::from_senders(tx);
        let runtime = runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        runtime.block_on(async {
            client
                .publish(
                    ValidatedTopic::new("hello/world").unwrap(),
                    QoS::ExactlyOnce,
                    false,
                    "good bye",
                )
                .await
                .expect("Should be able to publish");

            client
                .publish_with_properties(
                    ValidatedTopic::new("hello/world").unwrap(),
                    QoS::ExactlyOnce,
                    false,
                    "good bye",
                    PublishProperties::default(),
                )
                .await
                .expect("Should be able to publish");

            client
                .publish_bytes(
                    ValidatedTopic::new("hello/world").unwrap(),
                    QoS::ExactlyOnce,
                    false,
                    Bytes::from_static(b"good bye"),
                )
                .await
                .expect("Should be able to publish");

            client
                .publish_bytes_with_properties(
                    ValidatedTopic::new("hello/world").unwrap(),
                    QoS::ExactlyOnce,
                    false,
                    Bytes::from_static(b"good bye"),
                    PublishProperties::default(),
                )
                .await
                .expect("Should be able to publish");
        });

        let _ = rx.try_recv().expect("Should have message");
        let _ = rx.try_recv().expect("Should have message");
        let _ = rx.try_recv().expect("Should have message");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn async_try_publish_paths_accept_validated_topic() {
        let (tx, rx) = flume::bounded(4);
        let client = AsyncClient::from_senders(tx);

        client
            .try_publish(
                ValidatedTopic::new("hello/world").unwrap(),
                QoS::ExactlyOnce,
                false,
                "good bye",
            )
            .expect("Should be able to publish");

        client
            .try_publish_with_properties(
                ValidatedTopic::new("hello/world").unwrap(),
                QoS::ExactlyOnce,
                false,
                "good bye",
                PublishProperties::default(),
            )
            .expect("Should be able to publish");

        let _ = rx.try_recv().expect("Should have message");
        let _ = rx.try_recv().expect("Should have message");
    }

    #[test]
    fn async_publishing_invalid_raw_topic_fails() {
        let (tx, _) = flume::bounded(1);
        let client = AsyncClient::from_senders(tx);
        let runtime = runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        runtime.block_on(async {
            let err = client
                .publish("a/+/b", QoS::ExactlyOnce, false, "good bye")
                .await
                .expect_err("Invalid publish topic should fail");
            assert!(
                matches!(err, ClientError::Request(req) if matches!(*req, Request::Publish(_)))
            );

            let err = client
                .publish_bytes(
                    "a/+/b",
                    QoS::ExactlyOnce,
                    false,
                    Bytes::from_static(b"good bye"),
                )
                .await
                .expect_err("Invalid publish topic should fail");
            assert!(
                matches!(err, ClientError::Request(req) if matches!(*req, Request::Publish(_)))
            );

            let err = client
                .publish_with_properties(
                    "",
                    QoS::AtMostOnce,
                    false,
                    "good bye",
                    PublishProperties::default(),
                )
                .await
                .expect_err("Empty topic without topic alias should fail");
            assert!(
                matches!(err, ClientError::Request(req) if matches!(*req, Request::Publish(_)))
            );
        });
    }

    #[test]
    fn disconnect_with_properties_builds_disconnect_request() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let properties = DisconnectProperties {
            session_expiry_interval: Some(120),
            reason_string: Some("closing".to_string()),
            user_properties: vec![("source".to_string(), "test".to_string())],
            server_reference: Some("backup-broker".to_string()),
        };

        client
            .disconnect_with_properties(
                DisconnectReasonCode::ImplementationSpecificError,
                properties.clone(),
            )
            .expect("disconnect_with_properties should enqueue request");

        let request = rx.try_recv().expect("Should have disconnect request");
        match request {
            Request::Disconnect(disconnect) => {
                assert_eq!(
                    disconnect.reason_code,
                    DisconnectReasonCode::ImplementationSpecificError
                );
                assert_eq!(disconnect.properties, Some(properties));
            }
            request => panic!("Expected disconnect request, got {request:?}"),
        }
    }

    #[test]
    fn try_disconnect_with_properties_builds_disconnect_request() {
        let (tx, rx) = flume::bounded(1);
        let client = Client::from_sender(tx);
        let properties = DisconnectProperties {
            session_expiry_interval: Some(360),
            reason_string: Some("maintenance".to_string()),
            user_properties: vec![("env".to_string(), "test".to_string())],
            server_reference: None,
        };

        client
            .try_disconnect_with_properties(
                DisconnectReasonCode::ServerShuttingDown,
                properties.clone(),
            )
            .expect("try_disconnect_with_properties should enqueue request");

        let request = rx.try_recv().expect("Should have disconnect request");
        match request {
            Request::Disconnect(disconnect) => {
                assert_eq!(
                    disconnect.reason_code,
                    DisconnectReasonCode::ServerShuttingDown
                );
                assert_eq!(disconnect.properties, Some(properties));
            }
            request => panic!("Expected disconnect request, got {request:?}"),
        }
    }

    #[test]
    fn async_disconnect_with_properties_builds_disconnect_request() {
        let (tx, rx) = flume::bounded(1);
        let client = AsyncClient::from_senders(tx);
        let runtime = runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let properties = DisconnectProperties {
            session_expiry_interval: Some(42),
            reason_string: Some("done".to_string()),
            user_properties: vec![("k".to_string(), "v".to_string())],
            server_reference: Some("fallback".to_string()),
        };

        runtime.block_on(async {
            client
                .disconnect_with_properties(
                    DisconnectReasonCode::UseAnotherServer,
                    properties.clone(),
                )
                .await
                .expect("disconnect_with_properties should enqueue request");
        });

        let request = rx.try_recv().expect("Should have disconnect request");
        match request {
            Request::Disconnect(disconnect) => {
                assert_eq!(
                    disconnect.reason_code,
                    DisconnectReasonCode::UseAnotherServer
                );
                assert_eq!(disconnect.properties, Some(properties));
            }
            request => panic!("Expected disconnect request, got {request:?}"),
        }
    }

    #[test]
    fn async_try_disconnect_with_properties_builds_disconnect_request() {
        let (tx, rx) = flume::bounded(1);
        let client = AsyncClient::from_senders(tx);
        let properties = DisconnectProperties {
            session_expiry_interval: Some(7),
            reason_string: Some("bye".to_string()),
            user_properties: vec![("actor".to_string(), "test".to_string())],
            server_reference: None,
        };

        client
            .try_disconnect_with_properties(
                DisconnectReasonCode::AdministrativeAction,
                properties.clone(),
            )
            .expect("try_disconnect_with_properties should enqueue request");

        let request = rx.try_recv().expect("Should have disconnect request");
        match request {
            Request::Disconnect(disconnect) => {
                assert_eq!(
                    disconnect.reason_code,
                    DisconnectReasonCode::AdministrativeAction
                );
                assert_eq!(disconnect.properties, Some(properties));
            }
            request => panic!("Expected disconnect request, got {request:?}"),
        }
    }

    #[test]
    fn tracked_publish_requires_tracking_channel() {
        let (tx, _) = flume::bounded(2);
        let client = AsyncClient::from_senders(tx);
        let runtime = runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        runtime.block_on(async {
            let err = client
                .publish_tracked("hello/world", QoS::AtLeastOnce, false, "good bye")
                .await
                .expect_err("tracked publish should fail without tracked channel");
            assert!(matches!(err, ClientError::TrackingUnavailable));

            let err = client
                .publish_bytes_tracked(
                    "hello/world",
                    QoS::AtLeastOnce,
                    false,
                    Bytes::from_static(b"good bye"),
                )
                .await
                .expect_err("tracked publish bytes should fail without tracked channel");
            assert!(matches!(err, ClientError::TrackingUnavailable));

            let err = client
                .subscribe_tracked("hello/world", QoS::AtLeastOnce)
                .await
                .expect_err("tracked subscribe should fail without tracked channel");
            assert!(matches!(err, ClientError::TrackingUnavailable));

            let err = client
                .subscribe_many_tracked(vec![Filter::new("hello/world", QoS::AtLeastOnce)])
                .await
                .expect_err("tracked subscribe many should fail without tracked channel");
            assert!(matches!(err, ClientError::TrackingUnavailable));

            let err = client
                .unsubscribe_tracked("hello/world")
                .await
                .expect_err("tracked unsubscribe should fail without tracked channel");
            assert!(matches!(err, ClientError::TrackingUnavailable));
        });

        let err = client
            .try_subscribe_tracked("hello/world", QoS::AtLeastOnce)
            .expect_err("tracked try_subscribe should fail without tracked channel");
        assert!(matches!(err, ClientError::TrackingUnavailable));

        let err = client
            .try_subscribe_many_tracked(vec![Filter::new("hello/world", QoS::AtLeastOnce)])
            .expect_err("tracked try_subscribe_many should fail without tracked channel");
        assert!(matches!(err, ClientError::TrackingUnavailable));

        let err = client
            .try_unsubscribe_tracked("hello/world")
            .expect_err("tracked try_unsubscribe should fail without tracked channel");
        assert!(matches!(err, ClientError::TrackingUnavailable));
    }

    #[test]
    fn tracked_unsubscribe_uses_control_request_channel() {
        let (requests, requests_rx) = flume::bounded(1);
        let (control_requests, control_requests_rx) = flume::bounded(1);
        let (immediate_disconnect, _immediate_disconnect_rx) = flume::unbounded();
        let client = AsyncClient {
            request_tx: RequestSender::WithNotice {
                requests,
                control_requests,
                immediate_disconnect,
            },
        };
        let runtime = runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        runtime
            .block_on(client.unsubscribe_tracked("hello/world"))
            .expect("tracked unsubscribe should enqueue");

        assert!(requests_rx.is_empty());
        let envelope = control_requests_rx
            .try_recv()
            .expect("tracked unsubscribe should use control channel");
        assert!(matches!(envelope.into_parts().0, Request::Unsubscribe(_)));
    }

    #[test]
    fn tracked_auth_uses_control_request_channel() {
        let (requests, requests_rx) = flume::bounded(1);
        let (control_requests, control_requests_rx) = flume::bounded(1);
        let (immediate_disconnect, _immediate_disconnect_rx) = flume::unbounded();
        let client = AsyncClient {
            request_tx: RequestSender::WithNotice {
                requests,
                control_requests,
                immediate_disconnect,
            },
        };
        let runtime = runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        runtime
            .block_on(client.reauth_tracked(None))
            .expect("tracked auth should enqueue");

        assert!(requests_rx.is_empty());
        let envelope = control_requests_rx
            .try_recv()
            .expect("tracked auth should use control channel");
        assert!(matches!(envelope.into_parts().0, Request::Auth(_)));
    }
}