CloudAMQP with MQTT and Rust

We recommend rumqttc, the most widely used MQTT client for Rust. The examples use its blocking client; AsyncClient works the same way on Tokio.

cargo add rumqttc

Set MQTT_HOST, MQTT_PORT, MQTT_USERNAME and MQTT_PASSWORD to the MQTT details on your instance’s details page in the CloudAMQP console. Port 8883 uses TLS. Without them, the examples connect to a local broker.

Save the publisher as src/bin/publisher.rs:

use std::env;

use rumqttc::{Client, Event, MqttOptions, Packet, QoS, Transport};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let host = env::var("MQTT_HOST").unwrap_or("localhost".into());
    let port = env::var("MQTT_PORT").unwrap_or("1883".into()).parse()?;
    let mut options = MqttOptions::new(format!("publisher-{}", std::process::id()), host, port);
    options.set_credentials(
        env::var("MQTT_USERNAME").unwrap_or("guest".into()),
        env::var("MQTT_PASSWORD").unwrap_or("guest".into()),
    );
    if port == 8883 {
        options.set_transport(Transport::tls_with_default_config());
    }
    let (client, mut connection) = Client::new(options, 10);

    client.publish("hello", QoS::AtLeastOnce, false, "Hello world!")?;
    // The connection sends the message; QoS 1 waits for the broker's PUBACK
    for event in connection.iter() {
        if let Event::Incoming(Packet::PubAck(_)) = event? {
            println!(" [x] Sent 'Hello world!'");
            break;
        }
    }
    client.disconnect()?;
    Ok(())
}

Save the subscriber as src/bin/worker.rs:

use std::env;

use rumqttc::{Client, Event, MqttOptions, Packet, QoS, Transport};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let host = env::var("MQTT_HOST").unwrap_or("localhost".into());
    let port = env::var("MQTT_PORT").unwrap_or("1883".into()).parse()?;
    let mut options = MqttOptions::new(format!("worker-{}", std::process::id()), host, port);
    options.set_credentials(
        env::var("MQTT_USERNAME").unwrap_or("guest".into()),
        env::var("MQTT_PASSWORD").unwrap_or("guest".into()),
    );
    if port == 8883 {
        options.set_transport(Transport::tls_with_default_config());
    }
    let (client, mut connection) = Client::new(options, 10);

    client.subscribe("hello", QoS::AtLeastOnce)?;
    // rumqttc acks each QoS 1 message as it arrives
    for event in connection.iter() {
        match event? {
            Event::Incoming(Packet::SubAck(_)) => {
                println!(" [*] Waiting for messages. To exit press CTRL+C")
            }
            Event::Incoming(Packet::Publish(message)) => {
                println!(
                    " [x] Received '{}'",
                    String::from_utf8_lossy(&message.payload)
                )
            }
            _ => {}
        }
    }
    Ok(())
}

Run cargo run --bin worker in one terminal and cargo run --bin publisher in another.

A runnable example is in rust/rumqttc in cloudamqp/examples.

Good to know

Put the vhost in the username, as vhost:user. MQTT has no vhost of its own, and the console shows the username that way.
Transport::tls_with_default_config() turns on TLS, verified against the system’s certificates with rustls.
Iterate the connection. Nothing is sent or received until you poll connection.iter(); the publisher polls it until the broker’s PUBACK arrives.
Messages are acked as they arrive. rumqttc acks each QoS 1 message before the loop sees it; set set_manual_acks(true) to ack after processing.

← C

Command line →