Rust client library

mercurio-client is an async MQTT client for Tokio, speaking MQTT 3.1, 3.1.1 and 5.0, with automatic reconnection.

Adding it

[dependencies]
mercurio-client = "0.4"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

Features: tls, on by default, adds TLS through rustls and its aws-lc-rs provider; plain TCP and Unix socket users can turn it off with default-features = false to skip building rustls and aws-lc, and then connecting with any TLS setting fails with ClientError::Tls. websocket, off by default, adds MQTT over WebSocket. clap, off by default, adds the command-line arguments described below.

Connecting, publishing, subscribing

use mercurio_client::{ConnectOptions, Event, MqttClient, QoS};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let options = ConnectOptions::new("localhost", 1883).client_id("thermostat");
    let client = MqttClient::connect(options).await?;

    client.subscribe(&[("sensors/+/temperature", QoS::AtLeastOnce)]).await?;
    client.publish("sensors/hall/temperature", "20.5", QoS::AtLeastOnce).await?;

    while let Some(event) = client.recv().await {
        match event {
            Event::Message { topic, payload, .. } => {
                println!("{topic}: {}", String::from_utf8_lossy(&payload));
            }
            Event::Disconnected { reason } => eprintln!("disconnected: {reason:?}"),
            Event::Reconnected { .. } => eprintln!("reconnected"),
            Event::Stopped { error, .. } => eprintln!("gave up: {error}"),
            // Event and its variants are non_exhaustive: later versions
            // may add variants and fields
            _ => {}
        }
    }
    Ok(())
}

• publish and subscribe resolve once the broker acknowledged them: PUBACK, PUBCOMP or SUBACK.
• recv returns the next event, and None once the client is closed for good.
• Up to 256 events wait for recv. Once that many wait, the client stops reading new messages from the broker until the application takes some, but keeps sending pings and requests. A publish awaited inside the recv loop still completes, and no message is dropped.
• Dropping the last handle disconnects normally, as disconnect does: the broker discards the will.
• MqttClient methods take &self, so one client can be shared between tasks behind an Arc.

Connection options

ConnectOptions is a builder:

The client follows the limits the broker announces in CONNACK, such as Maximum QoS, Receive Maximum, Maximum Packet Size and Retain Available, and holds the broker to the client's own. client.connack() returns them as a ConnAckInfo: the keep-alive in effect, Maximum QoS, Retain Available, Receive Maximum, Maximum Packet Size, Topic Alias Maximum, the session expiry, the response information and the subscription features, each with its MQTT default when the broker left it out. It is replaced on every reconnection, before Event::Reconnected arrives. client.client_id() returns the client id in use, the one the broker assigned when the options had none.

The optional clap feature adds cli::ConnectArgs: the connection options recado and mercurioctl share, to flatten into your own clap command, and turned into ConnectOptions by to_connect_options().

Persistent sessions and reconnection

use std::time::Duration;
use mercurio_client::{ConnectOptions, MqttClient, Will};

let options = ConnectOptions::new("broker.example.com", 1883)
    .client_id("monitor")
    .clean_start(false)
    .session_expiry(3600)
    .reconnect(Duration::from_secs(5))
    .will(Will::new("status/monitor", "offline").retain(true));
let client = MqttClient::connect(options).await?;

When the connection drops, the client emits Event::Disconnected, retries, and emits Event::Reconnected { session_present }. The first attempt comes five to ten seconds later; each failed one doubles the wait, up to reconnect_max, and every wait is randomised by up to half, so clients that lost the same broker don't all come back at once. The client keeps trying while the broker answers Server Unavailable, Server Busy, Quota Exceeded or Connection Rate Exceeded. It stops for any other refusal, such as bad credentials or a ban, and for a redirection it doesn't follow (see follow_redirects): then Event::Stopped { error } is the last event before recv returns None. A write that takes longer than the write timeout, as on a half-open connection, also counts as a lost connection. Messages not yet acknowledged are sent again. QoS 1 and 2 publishes made while disconnected wait and are sent after reconnecting; QoS 0 ones fail with ClientError::Disconnected. Without session_present, the broker forgot the session and the subscriptions have to be made again: with resubscribe(true) the client does it before Event::Reconnected arrives, with the options and subscription identifiers they had, and reports each one the broker refuses this time with Event::ResubscribeFailed { topic, reason }, after Event::Reconnected. Unsubscribing forgets a subscription.

Manual acknowledgement

let options = ConnectOptions::new("broker.example.com", 1883)
    .manual_acks(true)
    .receive_maximum(100);
let client = MqttClient::connect(options).await?;
while let Some(event) = client.recv().await {
    if let Event::Message { payload, ack, .. } = event {
        store(&payload).await?;
        if let Some(token) = ack {
            client.ack(token).await?;
        }
    }
}

By default the client acknowledges QoS 1 and 2 messages as they arrive. With manual_acks(true), each Event::Message of QoS 1 or 2 carries an AckToken, and the PUBACK, or for QoS 2 the PUBREC, goes out only when the application passes it to ack. Until then the message counts against the client's receive_maximum, so an MQTT 5 broker sends no more than that many before some are acknowledged. Acknowledgements go out in the order the messages came: acknowledging a message waits for the earlier ones.

A message that is never acknowledged, for instance because its event was dropped, keeps its place under Receive Maximum and holds back the acknowledgements of later messages until the connection ends. Messages still unacknowledged then are forgotten, and their tokens do nothing; a broker that kept the session sends them again after the reconnection, as new events with new tokens. Acknowledging after processing therefore gives at-least-once delivery.

MQTT 5 properties

use mercurio_client::{MessageProperties, QoS};

let properties = MessageProperties::default()
    .content_type("application/json")
    .response_topic("replies/thermostat")
    .correlation_data("42")
    .message_expiry(60)
    .user_property("source", "hall");
client
    .publish_with_properties("requests/lamp", r#"{"on":true}"#, QoS::AtLeastOnce, false, properties)
    .await?;

Received messages carry their properties in Event::Message { properties, .. }, along with qos, retain and the subscription_ids they matched. MessageProperties, Will, SubscribeResult and UnsubscribeResult are #[non_exhaustive], as are the struct-like variants of Event, DisconnectReason and ClientError apart from Event::Disconnected: build them with their constructors and setters, and match them with .., so new fields don't break your code.

Request and response

use std::time::Duration;
use mercurio_client::RequestOptions;

let options = RequestOptions::new(Duration::from_secs(5)).response_prefix("replies");
let response = client.request("requests/lamp", r#"{"on":true}"#, options).await?;
println!("{}", String::from_utf8_lossy(&response.payload));

request publishes the message with a Response Topic and Correlation Data unique to the request, and resolves with the message a responder sends back with the same Correlation Data. The response topic is the broker's Response Information when request_response_information(true) got one, otherwise the prefix given, followed by a token of this client; the client subscribes to it once per session. Responses, late ones included, never come out of recv, so request and the event loop of the application work side by side. It fails with ClientError::Timeout when no response came in time, SubscribeRefused when the broker refuses the response topic, and as a publish does otherwise. MQTT 5 only.

Subscription options

use mercurio_client::{QoS, RetainHandling, SubscriptionOptions};

let options = SubscriptionOptions {
    qos: QoS::AtLeastOnce,
    no_local: true,
    retain_as_pub: true,
    retain_handling: RetainHandling::DoNotSendRetained,
};
// Messages matching it carry subscription id 7
let results = client.subscribe_with(&[("chat/#", options)], Some(7)).await?;
assert!(results[0].success);

unsubscribe(&["chat/#"]) removes subscriptions; both return a result per filter with the broker's reason code.

TLS

let options = ConnectOptions::new("iot.example.com", 8883)
    .ca_path("/etc/ssl/my-ca.pem")
    .client_cert("/etc/device/cert.pem", "/etc/device/key.pem");

A CA path, a client certificate, ALPN protocols or danger_skip_tls_verify(true) turns TLS on, so credentials never go out in the clear by mistake. tls(true) alone verifies the broker with the operating system's certificate store (on Linux the system CA bundle, such as /etc/ssl/certs; the Keychain on macOS; the certificate store on Windows), or, when that holds no usable certificate, the Mozilla roots bundled with the client; the client logs a warning when it falls back. With ca_path, the certificates in that file are the only ones trusted. A later tls(false) turns TLS off again. TLS isn't available over a Unix socket, and connect refuses these options there.

For full control, such as certificates held in memory, a verifier of your own or other TLS versions, build a rustls::ClientConfig, which the crate re-exports as mercurio_client::rustls, and pass it to tls_config. The broker's host name is still the name checked. It replaces the CA path, the client certificate, ALPN and skipped verification: setting any of them as well makes connect fail with InvalidArgument.

use std::sync::Arc;
use mercurio_client::rustls;

let mut roots = rustls::RootCertStore::empty();
roots.add_parsable_certificates(my_ca_certificates);
let mut config = rustls::ClientConfig::builder()
    .with_root_certificates(roots)
    .with_no_client_auth();
config.alpn_protocols = vec![b"mqtt".to_vec()];
let options = ConnectOptions::new("iot.example.com", 443).tls_config(Arc::new(config));

WebSocket

With the websocket feature, websocket(path) reaches brokers that serve MQTT over WebSocket, such as Mercurio's websocket listeners, or ones behind an HTTP proxy that only lets WebSocket through:

mercurio-client = { version = "0.4", features = ["websocket"] }

let options = ConnectOptions::new("broker.example.com", 443)
    .tls(true)
    .websocket("/mqtt");

The client asks for the mqtt subprotocol, sends packets in binary frames, and refuses a server that doesn't select mqtt. With TLS on, the connection is wss://, verified as any TLS connection. recado and mercurioctl don't offer WebSocket yet.

Enhanced authentication

SCRAM-SHA-256 is built in. The password never reaches the broker, and the client checks the broker's signature in the CONNACK, and after each reauthenticate(); without it the connection fails with ClientError::AuthenticationFailed. It needs MQTT 5, and a broker that offers the method, as mercuriod does:

use mercurio_client::{ConnectOptions, ScramSha256};

let options = ConnectOptions::new("localhost", 1883)
    .authenticator(ScramSha256::new("alice", "secret"));

For SCRAM-SHA-256-PLUS, which ties the login to the TLS connection (details), use ScramSha256::new("alice", "secret").channel_binding(true). It needs a TLS 1.3 connection made by the client itself: on any other connection, and with a custom connector, which never provides the TLS exporter, it fails before CONNECT is sent with "channel binding needs a TLS 1.3 connection".

Otherwise, implement Authenticator for a challenge-response method such as SCRAM. The exchange starts on every connection, reconnection and call to reauthenticate(). start() returns an error to stop the connection before CONNECT is sent. Before it, the client calls channel(exporter) with the 32-byte TLS exporter value of the connection, or None when it has none; the default does nothing. Authentication data is Bytes, which the crate re-exports from the bytes crate, as it does ReasonCode, QoS and ProtocolVersion:

use mercurio_client::{Authenticator, Bytes};

struct Token(String);

impl Authenticator for Token {
    fn method(&self) -> String {
        "TOKEN".into()
    }

    fn start(&mut self) -> Result<Option<Bytes>, String> {
        Ok(Some(Bytes::from(self.0.clone())))
    }

    fn respond(&mut self, _challenge: Option<Bytes>) -> Result<Option<Bytes>, String> {
        Err("TOKEN has no challenges".into())
    }
}

let options = ConnectOptions::new("localhost", 1883).authenticator(Token("secret".into()));

Errors and disconnections

Calls return ClientError, for example ConnectionRefused(reason), Redirected { server_reference, .. }, PublishRefused(reason) when the broker refuses a QoS 1 or 2 publish, InvalidArgument(..) for a topic name with wildcards, an empty or malformed filter list or a subscription identifier outside 1 to 268,435,455, which never reach the broker, Timeout, ConnectionFailed(io_error), Tls { .. }, Disconnected, or Closed once the client has stopped. ConnectionFailed keeps the std::io::Error, so a program can tell a refused connection from a missing socket by its kind, and Tls keeps the rustls or I/O error as its source. ClientError::is_transient() says whether trying again later may succeed: true for failed or lost connections and for refusals for want of capacity, false for invalid requests, refused credentials, redirections and TLS errors. Event::Disconnected says why the connection ended with a DisconnectReason: closed by the client, by the broker with a reason code, lost, or timed out.

disconnect() sends a normal DISCONNECT, after which the broker discards the will. With MQTT 5, disconnect_with(reason, session_expiry) sends another reason, such as ReasonCode::DisconnectWithWillMessage to have the will published anyway, and can change how long the broker keeps the session; it refuses reasons only a broker sends, and a non-zero session expiry when CONNECT asked for none, as MQTT forbids extending a session that was to end with the connection.

Examples in the repository

• recado/: the command-line client, built on this library.
• mercurioctl/: the administration tool, which runs the broker's commands with MQTT 5 request and response.
• mercurio-client/examples/bench.rs: a load generator with concurrent publishers and subscribers.