# CloudAMQP with MQTT and Rust

We recommend [rumqttc](https://github.com/bytebeamio/rumqtt/tree/main/rumqttc), the most widely used MQTT client for Rust. The examples use its blocking client; `AsyncClient` works the same way on Tokio.

```bash
cargo add rumqttc
```

## Publish and subscribe

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`:

```rust
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`:

```rust
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](https://github.com/cloudamqp/examples/tree/main/rust/rumqttc).

## 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.