CloudAMQP with Rust

We recommend amqp-client.rs, built on Tokio. It’s on crates.io as amqp-client. Create a project with this Cargo.toml (the package name is used by the entry points below):

[package]
name = "rust-amqp-client-example"
version = "0.1.0"
edition = "2024"
publish = false

[dependencies]
amqp-client = "0.1.0"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }

Publish and consume

Set CLOUDAMQP_URL to the AMQP URL from your instance’s details page in the CloudAMQP console. Without it, the example connects to a local broker.

Save the connection, publisher and consumer helpers as src/lib.rs:

use amqp_client::{BasicProperties, Client, Error, QueueOpts, RecoveryConfig, SubscribeOpts};

/// Connects to CLOUDAMQP_URL, and recovers the connection, queues and consumers if it's lost.
pub async fn connect() -> Result<Client, Error> {
    let url =
        std::env::var("CLOUDAMQP_URL").unwrap_or_else(|_| "amqp://guest:guest@localhost".into());
    Client::builder(&url)?
        .recovery(RecoveryConfig::default())
        .build()
        .await
}

/// Publishes a persistent message and returns once the broker has confirmed it.
pub async fn publish(client: &Client, queue: &str, message: &str) -> Result<(), Error> {
    client.queue(queue, QueueOpts::default()).await?;
    let props = BasicProperties::builder()
        .content_type("text/plain")
        .persistent()
        .build()?;
    client
        .publish_confirmed_with_props("", queue, &message.as_bytes().to_vec(), props)
        .await?;
    println!(" [x] Sent '{message}'");
    Ok(())
}

/// Calls `handler` for each message, with a prefetch of 10, and acks the message after it returns.
pub async fn consume(
    client: &Client,
    queue: &str,
    mut handler: impl FnMut(String),
) -> Result<(), Error> {
    let queue = client.queue(queue, QueueOpts::default()).await?;
    let mut consumer = queue
        .subscribe::<Vec<u8>>(SubscribeOpts {
            prefetch: 10,
            ..Default::default()
        })
        .await?;
    while let Some(delivery) = consumer.next().await {
        match delivery {
            Ok(d) => {
                handler(String::from_utf8_lossy(&d.body).into_owned());
                d.ack()?;
            }
            // Not decodable: dead-letter it instead of redelivering it forever
            Err(f) => f.reject(false)?,
        }
    }
    Ok(())
}

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

use rust_amqp_client_example::{connect, publish};

#[tokio::main]
async fn main() -> Result<(), amqp_client::Error> {
    let client = connect().await?;
    publish(&client, "hello", "Hello CloudAMQP!").await
}

Save a consumer as src/bin/worker.rs:

use rust_amqp_client_example::{connect, consume};

#[tokio::main]
async fn main() -> Result<(), amqp_client::Error> {
    let client = connect().await?;
    println!(" [*] Waiting for messages. To exit press CTRL+C");
    tokio::select! {
        result = consume(&client, "hello", |message| println!(" [x] Received '{message}'")) => result,
        _ = tokio::signal::ctrl_c() => Ok(()),
    }
}

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

A runnable example is in rust/amqp-client in cloudamqp/examples.

Good to know

Connect with the URL. Client::builder(&url)?, where the URL is the one on your instance’s details page in the CloudAMQP console. With amqps:// the server certificate is verified.
Turn on recovery with .recovery(RecoveryConfig::default()) on the builder. The client then reconnects and declares queues and consumers again after a connection loss. Without it, a lost connection is final.
Named queues are durable with QueueOpts::default(). Publish persistent messages explicitly, as in the example: BasicProperties defaults to transient.
Use publish_confirmed (or publish_confirmed_with_props, as in the example) to know that the broker has the message.

See the amqp-client.rs README for codecs and the low-level API.

Other Rust libraries

lapin, a widely used async client

← Go

.NET →