ruststream-rumqttc 0.7.0

//! The broker ladder: [`MqttBroker`] -> [`ConnectedMqttBroker`].
//!
//! Construction is synchronous and I/O-free; `connect` spawns the connection task and waits
//! for the broker's first `CONNACK`, and the connected form holds the live client directly.
//! One shared cell remains so publishers can be handed out while the application is still
//! being assembled, before `connect` runs.

// Without the `testing` feature a link has one variant, so a `match` on it has a single arm; the
// matches stay so that the in-process arm has its place when the feature is on.
#![cfg_attr(
    not(feature = "testing"),
    allow(clippy::infallible_destructuring_match)
)]

use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

use rumqttc::Transport;
#[cfg(feature = "testing")]
use rumqttc::v5::mqttbytes::matches;
use rumqttc::v5::mqttbytes::v5::{LastWill, Publish};
use rumqttc::v5::{AsyncClient, MqttOptions};
#[cfg(feature = "testing")]
use ruststream::testing::{Coordinator, InProcess, TestableBroker};
use ruststream::{
    AckError, AddressedCopies, Broker, ConnectedBroker, DefaultPublish, DescribeServer, ServerSpec,
    Subscribe,
};
#[cfg(feature = "testing")]
use ruststream::{OutgoingMessage, RawMessage};
use tokio::sync::{OnceCell, mpsc, oneshot};

#[cfg(feature = "asyncapi")]
use crate::asyncapi::{self, MqttLastWill, MqttServer};
use crate::conn::{Conn, Shared, run};
use crate::error::MqttError;
use crate::filter::{MqttFilter, MqttTopic, Qos};
#[cfg(feature = "testing")]
use crate::in_process::Bus;
use crate::publisher::{MqttPublish, MqttPublisher};
use crate::subscriber::MqttSubscriber;

/// What a connected broker and every handle derived from it speak over: the live client, or,
/// under the `testing` feature, the in-process transport the test harness connected instead.
///
/// Without the feature there is one variant, so the type is the client handle itself and every
/// `match` on it is irrefutable: a production build carries no second transport and no branch to
/// it.
#[derive(Clone)]
pub(crate) enum Link {
    Wire(AsyncClient),
    #[cfg(feature = "testing")]
    InProcess(Arc<Bus>),
}

// The zero-cost promise of the in-process mode, held by the compiler: a build without it gives
// the link exactly the size of the client it wraps, and every handle holding a link keeps its
// size with it.
/// How long shutdown waits for the connection task to write what was queued before it.
const SHUTDOWN_FLUSH: Duration = Duration::from_secs(5);

#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<Link>() == size_of::<AsyncClient>());
#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<Core>() == size_of::<(AsyncClient, Arc<Shared>)>());
#[cfg(not(feature = "testing"))]
const _: () =
    assert!(size_of::<Option<(Link, Publish)>>() == size_of::<Option<(AsyncClient, Publish)>>());

impl std::fmt::Debug for Link {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Link").finish_non_exhaustive()
    }
}

impl Link {
    /// Acknowledges the delivery `publish` to the server.
    pub(crate) async fn ack(&self, publish: &Publish) -> Result<(), AckError> {
        let acknowledged = match self {
            Self::Wire(client) => client.ack(publish).await.is_ok(),
            #[cfg(feature = "testing")]
            Self::InProcess(bus) => bus.is_open(),
        };
        if acknowledged {
            Ok(())
        } else {
            Err(AckError::Broker(Box::from(
                "the mqtt connection task has shut down",
            )))
        }
    }

    /// Closes the session with a clean `DISCONNECT`, and returns once the connection task has
    /// flushed it.
    async fn disconnect(&self, shared: &Shared) {
        match self {
            Self::Wire(client) => {
                let _ = client.disconnect().await;
                // An acknowledgement reports success once it is queued, so the session is closed
                // only when the task has written the queue up to the DISCONNECT, or has lost the
                // connection. The bound covers a broker that stops reading.
                if tokio::time::timeout(SHUTDOWN_FLUSH, shared.stopped())
                    .await
                    .is_err()
                {
                    tracing::warn!(
                        timeout = ?SHUTDOWN_FLUSH,
                        "mqtt shutdown stopped waiting for the connection task to flush its queue"
                    );
                }
            }
            #[cfg(feature = "testing")]
            Self::InProcess(bus) => bus.close(),
        }
    }
}

/// The connection state shared by the connected form and every handle derived from it.
pub(crate) struct Core {
    pub(crate) link: Link,
    pub(crate) shared: Arc<Shared>,
}

impl std::fmt::Debug for Core {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Core").finish_non_exhaustive()
    }
}

pub(crate) type CoreCell = Arc<OnceCell<Core>>;

/// An MQTT 5 broker for the `RustStream` messaging framework.
///
/// `new` is synchronous and records only configuration; the runtime connects once at startup
/// via the consuming [`Broker::connect`], which waits for the broker's `CONNACK`. That is
/// what lets a service compose with the synchronous `#[ruststream::app]` builder.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_rumqttc::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(MqttTopic::new("orders/created"))]
/// async fn confirm(order: &Order) -> HandlerOutcome {
///     println!("order {} confirmed", order.id);
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         MqttBroker::new("mqtt://localhost:1883", "orders-svc")
///             .credentials("user", "pass")
///             .keep_alive(Duration::from_secs(30))
///             .clean_start(false)
///             .session_expiry(Duration::from_hours(1)),
///         |b| {
///             b.include(confirm);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone)]
#[must_use]
pub struct MqttBroker {
    url: String,
    client_id: String,
    credentials: Option<(String, String)>,
    keep_alive: Option<Duration>,
    clean_start: Option<bool>,
    session_expiry: Option<u32>,
    max_packet_size: u32,
    receive_maximum: u16,
    last_will: Option<(String, Vec<u8>, Qos, bool)>,
    tls_ca: Option<Vec<u8>>,
    tls_client_auth: Option<(Vec<u8>, Vec<u8>)>,
    cell: CoreCell,
}

impl MqttBroker {
    /// Records the broker URL (`mqtt://host:port` or `mqtts://host:port`) and the client id.
    /// No I/O.
    pub fn new(url: impl Into<String>, client_id: impl Into<String>) -> Self {
        Self {
            url: url.into(),
            client_id: client_id.into(),
            credentials: None,
            keep_alive: None,
            clean_start: None,
            session_expiry: None,
            // The client's incoming default is 10 KiB, far below real payloads.
            max_packet_size: 1024 * 1024,
            receive_maximum: 1000,
            last_will: None,
            tls_ca: None,
            tls_client_auth: None,
            cell: Arc::new(OnceCell::new()),
        }
    }

    /// Username and password credentials.
    pub fn credentials(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
        self.credentials = Some((username.into(), password.into()));
        self
    }

    /// Sets the keep-alive interval (the protocol floor is 5 seconds).
    pub fn keep_alive(mut self, keep_alive: Duration) -> Self {
        self.keep_alive = Some(keep_alive);
        self
    }

    /// Starts a fresh session (`true`) or resumes a persistent one (`false`). Resuming pairs
    /// with [`session_expiry`](Self::session_expiry), and is what redelivers unacknowledged
    /// messages.
    pub fn clean_start(mut self, clean_start: bool) -> Self {
        self.clean_start = Some(clean_start);
        self
    }

    /// How long the broker keeps the session (and its subscriptions and unacked messages)
    /// after a disconnect.
    pub fn session_expiry(mut self, expiry: Duration) -> Self {
        self.session_expiry = Some(u32::try_from(expiry.as_secs()).unwrap_or(u32::MAX));
        self
    }

    /// The maximum incoming packet size. Defaults to 1 MiB (the client's own default is a
    /// 10 KiB cap that kills the connection on larger payloads).
    pub fn max_packet_size(mut self, bytes: u32) -> Self {
        self.max_packet_size = bytes;
        self
    }

    /// Flow control: how many unacknowledged `QoS` 1/2 deliveries the broker may have in
    /// flight, which is also what bounds an unread subscriber queue. Defaults to 1000.
    pub fn receive_maximum(mut self, maximum: u16) -> Self {
        self.receive_maximum = maximum;
        self
    }

    /// The last-will message the broker publishes if this session dies unexpectedly.
    pub fn last_will(
        mut self,
        topic: impl Into<String>,
        payload: impl Into<Vec<u8>>,
        qos: Qos,
        retain: bool,
    ) -> Self {
        self.last_will = Some((topic.into(), payload.into(), qos, retain));
        self
    }

    /// Trusts `ca` (PEM) for the TLS connection; selects TLS regardless of the URL scheme.
    ///
    /// The client verifies the server against this certificate authority and no other store, so
    /// it is what a TLS connection needs rather than an option on one: a `mqtts://` URL without
    /// it is refused when the broker connects, naming this setting.
    pub fn tls_ca(mut self, ca: impl Into<Vec<u8>>) -> Self {
        self.tls_ca = Some(ca.into());
        self
    }

    /// Authenticates with a TLS client certificate (both PEM), as managed MQTT services
    /// commonly require. Asks for TLS on its own, and [`tls_ca`](Self::tls_ca) is what verifies
    /// the server, so a connection that sets this and not that one is refused.
    pub fn tls_client_auth(mut self, cert: impl Into<Vec<u8>>, key: impl Into<Vec<u8>>) -> Self {
        self.tls_client_auth = Some((cert.into(), key.into()));
        self
    }

    /// A publisher sharing this broker's connection cell; buildable before `connect`.
    #[must_use]
    pub fn publisher(&self) -> MqttPublisher {
        MqttPublisher::new(Arc::clone(&self.cell), Qos::default(), false)
    }

    /// Whether the URL's scheme selects TLS.
    fn tls_from_scheme(&self) -> bool {
        self.url.starts_with("mqtts://")
    }

    /// The URL's authority: the host and port, without the scheme, the userinfo or anything after
    /// the host.
    ///
    /// An MQTT URL routinely carries `user:password@`, and that password must reach neither the
    /// connection nor the generated document. The framework owns that trimming, so the rule is
    /// the same one every broker crate applies.
    fn authority(&self) -> String {
        ServerSpec::host_from_url(&self.url)
    }

    /// The host and port a client connects to, and what the generated document reports.
    fn endpoint(&self) -> Result<(String, u16), MqttError> {
        let authority = self.authority();
        let (host, port) = match authority.rsplit_once(':') {
            Some((host, port)) => (
                host.to_owned(),
                port.parse::<u16>()
                    .map_err(|_| MqttError::Invalid(format!("'{port}' is not a valid port")))?,
            ),
            None => (
                authority.clone(),
                if self.tls_from_scheme() { 8883 } else { 1883 },
            ),
        };
        if host.is_empty() {
            return Err(MqttError::Invalid("host must be non-empty".into()));
        }
        Ok((host, port))
    }

    fn options(&self) -> Result<MqttOptions, MqttError> {
        let tls_from_scheme = self.tls_from_scheme();
        let (host, port) = self.endpoint()?;
        if let Some(keep_alive) = self.keep_alive
            && keep_alive < Duration::from_secs(5)
        {
            return Err(MqttError::Invalid(
                "keep_alive must be at least 5 seconds".into(),
            ));
        }

        let mut options = MqttOptions::new(self.client_id.clone(), host, port);
        if let Some(keep_alive) = self.keep_alive {
            options.set_keep_alive(keep_alive);
        }
        if let Some(clean_start) = self.clean_start {
            options.set_clean_start(clean_start);
        }
        if let Some((username, password)) = &self.credentials {
            options.set_credentials(username.clone(), password.clone());
        }
        if let Some(expiry) = self.session_expiry {
            options.set_session_expiry_interval(Some(expiry));
        }
        options.set_max_packet_size(Some(self.max_packet_size));
        options.set_receive_maximum(Some(self.receive_maximum));
        options.set_manual_acks(true);
        if let Some((topic, payload, qos, retain)) = &self.last_will {
            options.set_last_will(LastWill::new(
                topic.clone(),
                payload.clone(),
                qos.to_client(),
                *retain,
                None,
            ));
        }
        if tls_from_scheme || self.tls_ca.is_some() || self.tls_client_auth.is_some() {
            // The client verifies the server against the certificates handed to it here and
            // against no other store, so an empty one trusts nothing: every handshake would fail,
            // and the connection task would read that as transient and retry until `connect`
            // times out. Refusing here is the difference between a message naming the missing
            // setting and half a minute of silence.
            let Some(ca) = self.tls_ca.clone() else {
                return Err(MqttError::Invalid(
                    "a TLS connection needs the certificate authority to trust: call tls_ca(pem)"
                        .to_owned(),
                ));
            };
            options.set_transport(Transport::tls(ca, self.tls_client_auth.clone(), None));
        }
        Ok(options)
    }
}

impl Broker for MqttBroker {
    type Error = MqttError;
    type Connected = ConnectedMqttBroker;

    async fn connect(self) -> Result<Self::Connected, Self::Error> {
        let core = self
            .cell
            .get_or_try_init(async || {
                let options = self.options()?;
                let (client, eventloop) = AsyncClient::new(options, 64);
                let shared = Arc::new(Shared::new());
                let (connack_tx, connack_rx) = oneshot::channel();
                tokio::spawn(run(Conn {
                    client: client.clone(),
                    eventloop,
                    shared: Arc::clone(&shared),
                    first_connack: Some(connack_tx),
                }));
                // The task retries transient failures itself; the first CONNACK (or a fatal
                // refusal) decides whether connect succeeds.
                match tokio::time::timeout(Duration::from_secs(30), connack_rx).await {
                    Ok(Ok(Ok(()))) => {}
                    Ok(Ok(Err(err))) => return Err(err),
                    Ok(Err(_)) => {
                        return Err(MqttError::Connect(Box::from(
                            "the connection task exited before the first CONNACK",
                        )));
                    }
                    Err(_) => {
                        shared.closed.store(true, Ordering::Release);
                        // A task still dialing would otherwise connect later and run unowned;
                        // the queued DISCONNECT is what stops a task that got through.
                        let _ = client.try_disconnect();
                        // Everything the task retried is gone by now, so the wait is all there
                        // would be to report: a wrong port, an unreachable host and a handshake
                        // the broker refused after the fact would all read the same.
                        let reason = shared.last_error().map_or_else(
                            || "timed out waiting for the broker's CONNACK".to_owned(),
                            |last| {
                                format!(
                                    "timed out waiting for the broker's CONNACK; \
                                     last connection error: {last}"
                                )
                            },
                        );
                        return Err(MqttError::Connect(Box::from(reason)));
                    }
                }
                Ok::<_, MqttError>(Core {
                    link: Link::Wire(client),
                    shared,
                })
            })
            .await?;
        Ok(ConnectedMqttBroker {
            link: core.link.clone(),
            shared: Arc::clone(&core.shared),
            cell: self.cell,
        })
    }
}

/// The in-process mode: the connected form a test runs the production app against, carrying the
/// in-process transport in place of the client and every setting of this broker.
///
/// The options are built as `connect` builds them, so a configuration a service could not connect
/// with is not one a test can connect with either.
#[cfg(feature = "testing")]
impl InProcess for MqttBroker {
    async fn connect_in_process(self) -> Result<Self::Connected, Self::Error> {
        self.options()?;
        let max_packet_size = self.max_packet_size;
        let core = self
            .cell
            .get_or_try_init(async || {
                Ok::<_, MqttError>(Core {
                    link: Link::InProcess(Bus::new(max_packet_size)),
                    shared: Arc::new(Shared::new()),
                })
            })
            .await?;
        Ok(ConnectedMqttBroker {
            link: core.link.clone(),
            shared: Arc::clone(&core.shared),
            cell: self.cell,
        })
    }
}

#[cfg(feature = "testing")]
ruststream::register_testable_broker!(MqttBroker);

impl MqttBroker {
    /// The session this broker opens, as the `mqtt` server binding describes it.
    ///
    /// Credentials are absent by construction: the binding carries the client identity and the
    /// session settings, never the user name or the password. The last will contributes its
    /// coordinates and not its payload, which is content rather than a coordinate.
    #[cfg(feature = "asyncapi")]
    fn server_binding(&self) -> ruststream::asyncapi::Bindings {
        asyncapi::server(&MqttServer {
            client_id: self.client_id.clone(),
            clean_session: self.clean_start,
            last_will: self
                .last_will
                .as_ref()
                .map(|(topic, _payload, qos, retain)| MqttLastWill {
                    topic: topic.clone(),
                    qos: qos.level(),
                    retain: *retain,
                }),
            keep_alive: self.keep_alive.map(|interval| interval.as_secs()),
            session_expiry_interval: self.session_expiry,
            maximum_packet_size: self.max_packet_size,
        })
    }
}

impl DescribeServer for MqttBroker {
    /// Reports the host and port a client connects to, the protocol version it speaks, and the
    /// session settings the `mqtt` binding has room for. A URL's credentials stay out of the
    /// generated document, which teams publish and share.
    ///
    /// The port is stated even where the URL leaves it out, because the protocol's default is
    /// what a reader of the document would otherwise have to know.
    fn describe_server(&self) -> ServerSpec {
        // A URL `connect` will reject still must not hold up the document, so the fallback keeps
        // the framework's stripped authority and drops the unusable port.
        let spec = self.endpoint().map_or_else(
            |_| ServerSpec::from_url(&self.url, "mqtt"),
            |(host, port)| ServerSpec::new(format!("{host}:{port}"), "mqtt"),
        );
        // The crate speaks MQTT 5 and nothing else: user properties and shared subscriptions
        // exist only there.
        let spec = spec.protocol_version("5");
        #[cfg(feature = "asyncapi")]
        let spec = spec.bindings(self.server_binding());
        spec
    }
}

/// The typed witness that `connect` succeeded: holds the live client directly, or the in-process
/// transport when the test harness connected it.
pub struct ConnectedMqttBroker {
    link: Link,
    shared: Arc<Shared>,
    // Keeps the cell of publishers handed out before connect alive and filled.
    cell: CoreCell,
}

impl std::fmt::Debug for ConnectedMqttBroker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ConnectedMqttBroker")
            .finish_non_exhaustive()
    }
}

impl ConnectedMqttBroker {
    /// A publisher from the connected form with the default policy (`QoS` 1, not retained).
    #[must_use]
    pub fn publisher(&self) -> MqttPublisher {
        MqttPublisher::new(Arc::clone(&self.cell), Qos::default(), false)
    }

    pub(crate) fn publisher_with(&self, policy: MqttPublish) -> MqttPublisher {
        policy.into_publisher(Arc::clone(&self.cell))
    }

    /// Opens the subscription described by `topic` and waits for the broker's `SUBACK`.
    ///
    /// # Errors
    ///
    /// Returns [`MqttError`] when the value is not a topic, the broker rejects it, or the broker
    /// is shut down.
    pub async fn subscribe_topic(&self, topic: MqttTopic) -> Result<MqttSubscriber, MqttError> {
        topic.validate()?;
        self.open(topic.into_filter()).await
    }

    /// Opens the subscription described by `filter` and waits for the broker's `SUBACK`.
    ///
    /// # Errors
    ///
    /// Returns [`MqttError`] when the descriptor is invalid, the broker rejects the filter,
    /// or the broker is shut down.
    pub async fn subscribe_filter(&self, filter: MqttFilter) -> Result<MqttSubscriber, MqttError> {
        filter.validate()?;
        self.open(filter).await
    }

    /// The wire half both descriptors share, with validation already done.
    async fn open(&self, topic: MqttFilter) -> Result<MqttSubscriber, MqttError> {
        self.shared.ensure_open()?;

        let wire_filter = topic.wire_filter();
        let qos = topic.qos_value().to_client();
        let (tx, rx) = mpsc::unbounded_channel();
        let id = self
            .shared
            .open(&self.link, &wire_filter, topic.filter(), qos, tx)
            .await?;

        Ok(MqttSubscriber::new(
            topic.filter().to_owned(),
            id,
            Arc::clone(&self.shared),
            self.link.clone(),
            rx,
        ))
    }
}

impl ConnectedBroker for ConnectedMqttBroker {
    type Error = MqttError;
    type Closed = ();

    async fn shutdown(self) -> Result<(), Self::Error> {
        self.shared.closed.store(true, Ordering::Release);
        let held = self.shared.held();
        if held > 0 {
            // Unacknowledged, so the broker redelivers them when the session resumes; they are
            // lost only if the session is not persistent.
            tracing::warn!(
                held,
                "mqtt shutdown with deliveries no subscription ever matched"
            );
        }
        // A clean DISCONNECT lets the broker publish no last will and expire the session per
        // policy; the connection task exits once it has flushed it, after every acknowledgement
        // queued before it.
        self.link.disconnect(&self.shared).await;
        Ok(())
    }
}

impl Subscribe for ConnectedMqttBroker {
    type Subscriber = MqttSubscriber;
    /// A name written at the mount site is a topic, which is also the topic a publisher names, so
    /// a deferred retry reaches the subscription that took it. A filter with `+` or `#` is
    /// subscribe-only, and subscribing to one is what [`MqttFilter`] is for.
    type Copies = AddressedCopies;

    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error> {
        self.subscribe_topic(MqttTopic::new(name)).await
    }
}

impl DefaultPublish for ConnectedMqttBroker {
    type Policy = MqttPublish;
}

/// The harness's view of the in-process transport: what it injects, what it reads back, the
/// coordinator it counts the in-flight deliveries with, and which subscriptions a publish reaches.
///
/// # Panics
///
/// `inject` and `published` panic on a broker connected with `connect`: the harness drives only
/// the transport `connect_in_process` produced, and a live connection has no log to read and no
/// synchronous way to take a message.
#[cfg(feature = "testing")]
impl TestableBroker for ConnectedMqttBroker {
    fn install_coordinator(&self, coordinator: Coordinator) {
        if let Link::InProcess(bus) = &self.link {
            bus.install(coordinator);
        }
    }

    fn inject(&self, message: OutgoingMessage<'_>) {
        let bus = self.bus("inject");
        // An external producer is another client of the server, so its publish goes through the
        // server's routing like any other, at the default quality of service and not retained.
        if let Err(err) = bus.inject(&self.shared, &message) {
            panic!(
                "the injected message to {:?} is not one a server takes: {err}",
                message.name()
            );
        }
    }

    fn published(&self, name: &str) -> Vec<RawMessage> {
        self.bus("published").published(name)
    }

    /// Every subscription whose topic filter matches the destination, by the match the
    /// connection demultiplexes deliveries with; the subscriptions this connection opened on one
    /// wire filter (the members of one share group, or one filter subscribed twice) are one
    /// subscription to the server and take one delivery between them in turn, so the answer names
    /// the one whose turn it is.
    fn routes(&self, destination: &str, subscriptions: &[&str]) -> Vec<usize> {
        let mut routed = Vec::new();
        let mut answered: Vec<&str> = Vec::new();
        for filter in subscriptions {
            if !matches(destination, filter) || answered.contains(filter) {
                continue;
            }
            answered.push(filter);
            // The subscriptions the app opened under this name, in the order it opened them.
            let positions = subscriptions
                .iter()
                .enumerate()
                .filter(|(_, name)| *name == filter)
                .map(|(position, _)| position);
            let takers = self.shared.next_takers(filter);
            if takers.is_empty() {
                // A filter this connection has not opened answers by its name alone.
                routed.extend(positions);
            } else {
                let positions: Vec<usize> = positions.collect();
                routed.extend(takers.iter().filter_map(|rank| positions.get(*rank)));
            }
        }
        routed.sort_unstable();
        routed
    }
}

#[cfg(feature = "testing")]
impl ConnectedMqttBroker {
    /// The in-process transport, which is all the harness drives.
    fn bus(&self, what: &str) -> &Arc<Bus> {
        match &self.link {
            Link::InProcess(bus) => bus,
            Link::Wire(_) => panic!(
                "TestableBroker::{what} reached a broker connected with `connect`; the harness \
                 drives the connection `connect_in_process` produces"
            ),
        }
    }
}

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

    fn broker(url: &str) -> MqttBroker {
        MqttBroker::new(url, "describe")
    }

    #[test]
    fn the_endpoint_is_the_host_and_port_whatever_else_the_url_carries() {
        for (url, expected) in [
            ("mqtt://broker.example.com", ("broker.example.com", 1883)),
            ("mqtts://broker.example.com", ("broker.example.com", 8883)),
            (
                "mqtt://broker.example.com:1884",
                ("broker.example.com", 1884),
            ),
            (
                "mqtt://alice:s3cret@broker.example.com:1884",
                ("broker.example.com", 1884),
            ),
            (
                "mqtts://alice:s3cret@broker.example.com",
                ("broker.example.com", 8883),
            ),
            // A password may contain the separator, so the split takes the last one.
            (
                "mqtt://alice:p@ss@broker.example.com:1884",
                ("broker.example.com", 1884),
            ),
            (
                "mqtt://broker.example.com:1884/ns",
                ("broker.example.com", 1884),
            ),
            ("mqtt://broker.example.com/ns", ("broker.example.com", 1883)),
            // The authority ends before the path, so an `@` after it is not a separator.
            (
                "mqtt://broker.example.com/a@b",
                ("broker.example.com", 1883),
            ),
        ] {
            let broker = broker(url);
            let (host, port) = broker.endpoint().expect("the url parses");
            assert_eq!((host.as_str(), port), expected, "parsing {url}");
        }
    }

    /// TLS with nothing to verify the server against is a connection that can never succeed, so
    /// the setting that is missing is named instead. Each of the three ways to ask for TLS
    /// answers the same.
    #[tokio::test]
    async fn tls_without_a_certificate_authority_is_refused_before_any_io() {
        let scheme = MqttBroker::new("mqtts://127.0.0.1:8883", "tls-scheme");
        let client_auth =
            broker("mqtt://127.0.0.1:1").tls_client_auth(b"cert".to_vec(), b"key".to_vec());

        for (broker, asked) in [
            (scheme, "the URL scheme"),
            (client_auth, "a client certificate"),
        ] {
            let error = broker
                .connect()
                .await
                .expect_err("nothing can verify the server");
            assert!(
                matches!(&error, MqttError::Invalid(reason) if reason.contains("tls_ca")),
                "{asked} asks for TLS, and the refusal names what is missing: {error}"
            );
        }
    }

    /// The floor is the protocol's, and a value below it is refused where the service can still
    /// see it: before any I/O, with the number named. A broker that took it would negotiate a
    /// keep-alive nobody asked for.
    #[tokio::test]
    async fn a_keep_alive_below_the_protocol_floor_is_refused_before_any_io() {
        let error = broker("mqtt://127.0.0.1:1")
            .keep_alive(Duration::from_secs(1))
            .connect()
            .await
            .expect_err("the protocol floor is five seconds");

        assert!(
            matches!(&error, MqttError::Invalid(reason) if reason.contains("5 seconds")),
            "the refusal names the floor: {error}"
        );
    }

    #[test]
    fn a_url_with_credentials_describes_a_server_without_them() {
        let spec = broker("mqtt://alice:s3cret@broker.example.com:1884").describe_server();
        let host = spec.host.expect("a networked broker states its host");

        assert_eq!(host, "broker.example.com:1884");
        assert!(
            !host.contains('@'),
            "the userinfo separator is gone: {host}"
        );
        assert!(!host.contains("alice"), "the user name is gone: {host}");
        assert!(!host.contains("s3cret"), "the password is gone: {host}");
    }
}