Skip to main content

Crate rumqttc

Crate rumqttc 

Source
Expand description

§rumqttc-v4-next

crates.io page docs.rs page

rumqttc-v4-next is the explicit MQTT 3.1.1 client crate in the rumqtt family. Use it when you need MQTT 3.1.1 specifically. If you want the default MQTT 5 client, use rumqttc-next or rumqttc-v5-next instead.

This crate keeps the library target name as rumqttc so application code can stay familiar after migrating package names.

§Installation

cargo add rumqttc-v4-next

§Examples

§A simple synchronous publish and subscribe

use rumqttc::{MqttOptions, Client, QoS};
use std::time::Duration;
use std::thread;

let mut mqttoptions = MqttOptions::new("rumqtt-sync", "test.mosquitto.org");
mqttoptions.set_keep_alive(5);

let (mut client, mut connection) = Client::builder(mqttoptions).capacity(10).build();
client.subscribe("hello/rumqtt", QoS::AtMostOnce).unwrap();
thread::spawn(move || for i in 0..10 {
   client.publish("hello/rumqtt", QoS::AtLeastOnce, false, vec![i; i as usize]).unwrap();
   thread::sleep(Duration::from_millis(100));
});

// Iterate to poll the eventloop for connection progress
for (i, notification) in connection.iter().enumerate() {
    println!("Notification = {:?}", notification);
}

§A simple asynchronous publish and subscribe

use rumqttc::{MqttOptions, AsyncClient, QoS};
use tokio::{task, time};
use std::time::Duration;

#[tokio::main(flavor = "current_thread")]
async fn main() {
let mut mqttoptions = MqttOptions::new("rumqtt-async", "test.mosquitto.org");
mqttoptions.set_keep_alive(5);

let (mut client, mut eventloop) = AsyncClient::builder(mqttoptions).capacity(10).build();
client.subscribe("hello/rumqtt", QoS::AtMostOnce).await.unwrap();

task::spawn(async move {
    for i in 0..10 {
        client.publish("hello/rumqtt", QoS::AtLeastOnce, false, vec![i; i as usize]).await.unwrap();
        time::sleep(Duration::from_millis(100)).await;
    }
});

while let Ok(notification) = eventloop.poll().await {
    println!("Received = {:?}", notification);
}
}

§Tracked protocol results

Tracked APIs return notices that resolve with MQTT protocol responses: publish_tracked(...).wait_async() returns PublishResult, subscribe_tracked(...).wait_async() returns SubAck, and unsubscribe_tracked(...).wait_async() returns UnsubAck. Use wait_completion_async() when only completion/failure status is needed.

See examples/tracked_notices.rs for a runnable async example.

Quick overview of features

  • Eventloop orchestrates outgoing/incoming packets concurrently and handles the state
  • Pings the broker when necessary and detects client side half open connections as well
  • Throttling of outgoing packets (todo)
  • Queue size based flow control on outgoing packets
  • Automatic reconnections by just continuing the eventloop.poll()/connection.iter() loop
  • Natural backpressure to client APIs during bad network
  • Support for WebSockets
  • Secure transport using TLS
  • Unix domain sockets on Unix targets
  • Strict MQTT 3.1.1 packet validation on the codec path

In short, everything necessary to maintain a robust connection

Since the eventloop is externally polled (with iter()/poll() in a loop) out side the library and Eventloop is accessible, users can

  • Distribute incoming messages based on topics
  • Stop it when required
  • Access internal state for use cases like graceful shutdown or to modify options before reconnection

§Important notes

  • Looping on connection.iter()/eventloop.poll() is necessary to run the event loop and make progress. It yields incoming and outgoing activity notifications which allows customization as you see fit.

  • Blocking inside the connection.iter()/eventloop.poll() loop will block connection progress.

  • Bounded clients apply backpressure through the client request channel. If the same task that drives eventloop.poll() awaits publish(), subscribe(), unsubscribe(), ack(), or another request-sending API while that bounded channel is full, it can self-block: the request is waiting for the event loop to read from the channel, but the event loop cannot make progress until that task polls it again. For bounded async clients, prefer driving eventloop.poll() in a dedicated task and dispatch application work to other tasks. When dropping outgoing publishes under overload is intended, use try_publish() and treat a full-channel error as the drop signal.

  • Use client.disconnect()/try_disconnect() for MQTT-level graceful shutdown. The event loop first flushes previously accepted QoS 0 publishes and drains previously accepted QoS 1/ QoS 2 publish and tracked subscribe/unsubscribe state (SUBACK/UNSUBACK), then sends terminal DISCONNECT. Use disconnect_with_timeout() to bound this drain; if the timeout expires first, polling returns ConnectionError::DisconnectTimeout and DISCONNECT is not sent. Use disconnect_now() to send DISCONNECT immediately and abandon unresolved in-flight work. Dropping or aborting the transport closes locally without sending DISCONNECT and may trigger Will publication.

  • Disconnect requests use the normal client request channel. If the event loop is not currently reading new requests because the outbound QoS 1/ QoS 2 publish inflight window is full or a packet-id collision is pending, a queued disconnect_with_timeout() timeout starts only after the event loop observes the disconnect request, and disconnect_now() is not an out-of-band transport abort.

  • This crate is intentionally protocol-specific. It does not expose MQTT 5 features such as AUTH packets, topic aliases, or MQTT 5 property handling.

  • On Unix targets, local broker sockets are supported via MqttOptions::new(..., Broker::unix(...)). When the url feature is enabled, MqttOptions::parse_url("unix:///tmp/mqtt.sock?client_id=...") is also supported.

§TLS Support

rumqttc supports two TLS backends:

  • use-rustls (default): Uses rustls with aws-lc as the crypto provider and native platform certificates
  • use-native-tls: Uses the platform’s native TLS implementation (Secure Transport on macOS, SChannel on Windows, OpenSSL on Linux)

§TLS Feature Flags

FeatureDescription
use-rustlsEnable rustls with aws-lc provider (recommended default)
use-rustls-no-providerEnable rustls without selecting a crypto provider
use-rustls-aws-lcEnable rustls with aws-lc provider
use-rustls-ringEnable rustls with ring provider
use-native-tlsEnable native-tls backend

use-rustls-aws-lc and use-rustls-ring are mutually exclusive. Enabling both results in a compile error.

§Important: Using Both TLS Features

When both use-rustls-no-provider and use-native-tls features are enabled:

  • Configure TLS explicitly with MqttOptions::set_transport(Transport::tls_with_config(...))
  • Configure secure websockets explicitly with Broker::websocket("ws://...") plus MqttOptions::set_transport(Transport::wss_with_config(...))

Secure websocket connections upgrade the TCP stream using the selected TlsConfiguration before the websocket handshake, so backend selection follows the provided TLS configuration.

In dual-backend dependency graphs, avoid relying on TlsConfiguration::default(), because default backend selection must be explicit. Use TlsConfiguration::default_rustls() or TlsConfiguration::default_native() (when available) or pass an explicit configuration to Transport::tls_with_config(...) / Transport::wss_with_config(...).

Native-tls WSS can use platform roots via TlsConfiguration::default_native() or a custom CA / identity via TlsConfiguration::simple_native(...).

Re-exports§

pub use tokio_native_tls;use-native-tls
pub use tokio_rustls;use-rustls-no-provider
pub use mqttbytes::v4::*;
pub use mqttbytes::*;

Modules§

mqttbytes
mqttbytes

Structs§

AsyncClient
An asynchronous client, communicates with MQTT EventLoop.
AsyncClientBuilder
Builder for asynchronous MQTT clients.
Broker
Broker target used to construct MqttOptions.
Client
A synchronous client, communicates with MQTT EventLoop.
ClientBuilder
Builder for synchronous MQTT clients.
Connection
MQTT connection. Maintains all the necessary state
EventLoop
Eventloop with all the state of a connection
InvalidTopic
An error returned when a topic string fails validation against the MQTT specification.
Iter
Iterator which polls the EventLoop for connection progress
MqttOptions
Options to configure the behaviour of MQTT connection
MqttOptionsBuilder
Builder for MqttOptions.
MqttState
State of the mqtt connection.
MqttStateBuilder
Builder for low-level MQTT 3.1.1 protocol state.
NetworkOptions
Provides a way to configure low level network connection configurations
Proxyproxy
PublishNotice
Wait handle returned by tracked publish APIs.
RecvError
Error type returned by Connection::recv
SubscribeNotice
Wait handle returned by tracked subscribe APIs.
UnsubscribeNotice
Wait handle returned by tracked unsubscribe APIs.
ValidatedTopic
A newtype wrapper that guarantees its inner String is a valid MQTT topic.

Enums§

ClientError
Client Error
ConnectionError
Critical errors during eventloop polling
Event
Events which can be yielded by the event loop
ManualAck
Prepared acknowledgement packet for manual acknowledgement mode.
NoticeFailureReason
OptionError
Outgoing
Current outgoing activity on the eventloop
ProxyAuthproxy
ProxyTypeproxy
PublishNoticeError
PublishResult
PublishTopic
Topic argument accepted by publish APIs.
RecvTimeoutError
Error type returned by Connection::recv_timeout
Request
Requests by the client to mqtt event loop. Request are handled one by one.
StateError
Errors during state handling
SubscribeNoticeError
TlsConfigurationuse-rustls-no-provider or use-native-tls
TLS configuration method
TlsErroruse-rustls-no-provider or use-native-tls
Transport
Transport methods. Defaults to TCP.
TryRecvError
Error type returned by Connection::try_recv
UnsubscribeNoticeError

Functions§

default_socket_connect
Default TCP socket connection logic used by the MQTT event loop.

Type Aliases§

Incoming