Tutorials

These tutorials follow RabbitMQ's classic tutorials. Each pattern has a complete program that sends messages and receives them, so you can run it as is. CI runs every program on this page against FireMQ, so the code always works.

Start a broker first:

firemq

The Python programs use pika and the JavaScript programs use amqplib; both talk to FireMQ over AMQP 0-9-1 exactly as they would to RabbitMQ. The Rust programs use FireMQ's own SDK, firemq-client, over the FireMQ protocol. Set AMQP_URL or FIREMQ_URL if your broker is not on localhost.

Hello world

A producer sends a message to a queue and a consumer receives it. The empty exchange name "" is the default exchange, which delivers to the queue named by the routing key.

Python:

"""Hello world: send one message to a queue and receive it."""
import os

import pika

connection = pika.BlockingConnection(pika.URLParameters(os.environ.get("AMQP_URL", "amqp://localhost")))
channel = connection.channel()
channel.queue_declare(queue="hello")

channel.basic_publish(exchange="", routing_key="hello", body=b"Hello World!")
print(" [x] Sent 'Hello World!'")

for method, properties, body in channel.consume("hello", auto_ack=True, inactivity_timeout=5):
    print(f" [x] Received {body.decode()!r}")
    break
connection.close()

JavaScript:

// Hello world: send one message to a queue and receive it.
import amqp from "amqplib";

const conn = await amqp.connect(process.env.AMQP_URL ?? "amqp://localhost");
const ch = await conn.createChannel();
await ch.assertQueue("hello", { durable: false });

ch.sendToQueue("hello", Buffer.from("Hello World!"));
console.log(" [x] Sent 'Hello World!'");

await new Promise((resolve) =>
  ch.consume("hello", (msg) => {
    console.log(` [x] Received '${msg.content}'`);
    resolve();
  }, { noAck: true }),
);
await conn.close();

Rust:

//! Publish and consume one message.
//!
//! Start a broker with `firemq`, then run `cargo run -p firemq-client --example hello`.

use firemq_client::{Client, QueueOptions, SubscribeOptions};

#[tokio::main]
async fn main() -> Result<(), firemq_client::Error> {
    let url = std::env::var("FIREMQ_URL").unwrap_or_else(|_| "fire://localhost".into());
    let client = Client::connect(&url).await?;
    client
        .declare_queue("hello.rust", QueueOptions::default())
        .await?;
    client.publish("", "hello.rust", "Hello World!").await?;
    println!(" [x] Sent 'Hello World!'");

    let mut sub = client
        .subscribe("hello.rust", SubscribeOptions::default())
        .await?;
    if let Some(msg) = sub.next().await {
        println!(" [x] Received '{}'", msg.body_str().unwrap_or_default());
        msg.ack()?;
    }
    client.close().await
}

Work queues

Several workers share the tasks of one queue. Two settings make this safe and fair:

• Workers acknowledge a task only after finishing it, so a crashed worker's task goes to another worker.
• A prefetch of 1 gives each worker one unacknowledged task at a time, so a busy worker does not hoard tasks.

The queue is durable and the messages are persistent, so tasks survive a broker restart.

Python:

"""Work queues: share tasks between workers, one unacknowledged task each."""
import os

import pika

connection = pika.BlockingConnection(pika.URLParameters(os.environ.get("AMQP_URL", "amqp://localhost")))
channel = connection.channel()
channel.queue_declare(queue="task_queue", durable=True)

for n in range(1, 4):
    channel.basic_publish(
        exchange="",
        routing_key="task_queue",
        body=f"task {n}".encode(),
        properties=pika.BasicProperties(delivery_mode=pika.DeliveryMode.Persistent),
    )

channel.basic_qos(prefetch_count=1)  # at most one unacknowledged task per worker
done = 0
for method, properties, body in channel.consume("task_queue", inactivity_timeout=5):
    print(f" [x] Done {body.decode()}")
    channel.basic_ack(method.delivery_tag)
    done += 1
    if done == 3:
        break
connection.close()

JavaScript:

// Work queues: share tasks between workers, one unacknowledged task each.
import amqp from "amqplib";

const conn = await amqp.connect(process.env.AMQP_URL ?? "amqp://localhost");
const ch = await conn.createChannel();
await ch.assertQueue("task_queue", { durable: true });
for (let n = 1; n <= 3; n++) ch.sendToQueue("task_queue", Buffer.from(`task ${n}`), { persistent: true });

await ch.prefetch(1); // at most one unacknowledged task per worker
let done = 0;
await new Promise((resolve) =>
  ch.consume("task_queue", (msg) => {
    console.log(` [x] Done ${msg.content}`);
    ch.ack(msg);
    if (++done === 3) resolve();
  }),
);
await conn.close();

Rust (the SDK makes queues durable and messages persistent by default):

//! Work queues: share tasks between workers, a few unacknowledged tasks each.

use firemq_client::{Client, QueueOptions, SubscribeOptions};

#[tokio::main]
async fn main() -> Result<(), firemq_client::Error> {
    let url = std::env::var("FIREMQ_URL").unwrap_or_else(|_| "fire://localhost".into());
    let client = Client::connect(&url).await?;
    client
        .declare_queue("task_queue.rust", QueueOptions::default())
        .await?;
    for n in 1..=3 {
        // Resolves once the broker has written the task to disk.
        client
            .publish("", "task_queue.rust", format!("task {n}"))
            .await?;
    }

    let opts = SubscribeOptions {
        prefetch: 1,
        ..Default::default()
    };
    let mut tasks = client.subscribe("task_queue.rust", opts).await?;
    for _ in 0..3 {
        let Some(task) = tasks.next().await else {
            break;
        };
        println!(" [x] Done {}", task.body_str().unwrap_or_default());
        task.ack()?;
    }
    client.close().await
}

Publish/subscribe

A fanout exchange copies every message to every queue bound to it. Each subscriber declares its own temporary queue: an empty name asks the broker for a unique one, and exclusive deletes it when the connection closes.

Python:

"""Publish/subscribe: a fanout exchange copies each message to every bound queue."""
import os

import pika

connection = pika.BlockingConnection(pika.URLParameters(os.environ.get("AMQP_URL", "amqp://localhost")))
channel = connection.channel()
channel.exchange_declare(exchange="logs", exchange_type="fanout")

# Each subscriber gets its own temporary queue.
queues = [channel.queue_declare(queue="", exclusive=True).method.queue for _ in range(2)]
for q in queues:
    channel.queue_bind(exchange="logs", queue=q)

channel.basic_publish(exchange="logs", routing_key="", body=b"info: hello subscribers")
for q in queues:
    for method, properties, body in channel.consume(q, auto_ack=True, inactivity_timeout=5):
        print(f" [{q[-6:]}] {body.decode()}")
        break
    channel.cancel()  # stop this consumer before reading the next queue
connection.close()

JavaScript:

// Publish/subscribe: a fanout exchange copies each message to every bound queue.
import amqp from "amqplib";

const conn = await amqp.connect(process.env.AMQP_URL ?? "amqp://localhost");
const ch = await conn.createChannel();
await ch.assertExchange("logs", "fanout", { durable: false });
const queues = [];
for (let i = 0; i < 2; i++) {
  const { queue } = await ch.assertQueue("", { exclusive: true });
  await ch.bindQueue(queue, "logs", "");
  queues.push(queue);
}
ch.publish("logs", "", Buffer.from("info: hello subscribers"));
for (const q of queues) {
  await new Promise((resolve) => ch.consume(q, (msg) => { console.log(` [${q.slice(-6)}] ${msg.content}`); resolve(); }, { noAck: true }));
}
await conn.close();

Rust:

//! Publish/subscribe: a fanout exchange copies every message to every bound queue.

use firemq_client::{Client, ExchangeOptions, QueueOptions, SubscribeOptions};

#[tokio::main]
async fn main() -> Result<(), firemq_client::Error> {
    let url = std::env::var("FIREMQ_URL").unwrap_or_else(|_| "fire://localhost".into());
    let client = Client::connect(&url).await?;
    client
        .declare_exchange("logs.rust", "fanout", ExchangeOptions::default())
        .await?;

    // Two subscribers, each with its own temporary queue.
    let mut subs = Vec::new();
    for _ in 0..2 {
        let queue = client.declare_queue("", QueueOptions::temporary()).await?;
        client.bind_queue(&queue.name, "logs.rust", "").await?;
        subs.push(
            client
                .subscribe(&queue.name, SubscribeOptions::default())
                .await?,
        );
    }

    // The routing key is ignored by fanout exchanges.
    client.publish("logs.rust", "", "info: hello").await?;

    for (n, sub) in subs.iter_mut().enumerate() {
        let Some(msg) = sub.next().await else { break };
        println!(" [subscriber {n}] {}", msg.body_str().unwrap_or_default());
        msg.ack()?;
    }
    client.close().await
}

Routing

A direct exchange delivers a message to the queues bound with exactly its routing key. Here only the error message reaches the queue.

Python:

"""Routing: a direct exchange delivers by exact routing key."""
import os

import pika

connection = pika.BlockingConnection(pika.URLParameters(os.environ.get("AMQP_URL", "amqp://localhost")))
channel = connection.channel()
channel.exchange_declare(exchange="direct_logs", exchange_type="direct")
errors = channel.queue_declare(queue="", exclusive=True).method.queue
channel.queue_bind(exchange="direct_logs", queue=errors, routing_key="error")

for severity in ("info", "warning", "error"):
    channel.basic_publish(exchange="direct_logs", routing_key=severity, body=f"{severity} event".encode())

for method, properties, body in channel.consume(errors, auto_ack=True, inactivity_timeout=5):
    print(f" [x] {method.routing_key}: {body.decode()}")  # only the error event arrives
    break
connection.close()

Rust, with a queue bound to two keys:

//! Routing: a direct exchange delivers a message to the queues bound with its exact key.

use firemq_client::{Client, ExchangeOptions, QueueOptions, SubscribeOptions};

#[tokio::main]
async fn main() -> Result<(), firemq_client::Error> {
    let url = std::env::var("FIREMQ_URL").unwrap_or_else(|_| "fire://localhost".into());
    let client = Client::connect(&url).await?;
    client
        .declare_exchange("direct_logs.rust", "direct", ExchangeOptions::default())
        .await?;
    let queue = client.declare_queue("", QueueOptions::temporary()).await?;
    for severity in ["warning", "error"] {
        client
            .bind_queue(&queue.name, "direct_logs.rust", severity)
            .await?;
    }

    for (severity, text) in [
        ("info", "disk 40% full"),
        ("warning", "disk 85% full"),
        ("error", "disk full"),
    ] {
        client
            .publish("direct_logs.rust", severity, format!("{severity}: {text}"))
            .await?;
    }

    // The info message matched no binding, so only two arrive.
    let mut sub = client
        .subscribe(&queue.name, SubscribeOptions::default())
        .await?;
    for _ in 0..2 {
        let Some(msg) = sub.next().await else { break };
        println!(" [x] {}", msg.body_str().unwrap_or_default());
        msg.ack()?;
    }
    client.close().await
}

Topics

A topic exchange matches routing keys made of dot-separated words against binding patterns. * matches exactly one word and # matches zero or more words, so *.critical matches kern.critical but not kern.info.

Python:

"""Topics: route with word patterns; * matches one word, # matches zero or more."""
import os

import pika

connection = pika.BlockingConnection(pika.URLParameters(os.environ.get("AMQP_URL", "amqp://localhost")))
channel = connection.channel()
channel.exchange_declare(exchange="topic_logs", exchange_type="topic")
critical = channel.queue_declare(queue="", exclusive=True).method.queue
channel.queue_bind(exchange="topic_logs", queue=critical, routing_key="*.critical")

for key in ("kern.critical", "kern.info", "app.critical"):
    channel.basic_publish(exchange="topic_logs", routing_key=key, body=key.encode())

received = 0
for method, properties, body in channel.consume(critical, auto_ack=True, inactivity_timeout=5):
    print(f" [x] {body.decode()}")
    received += 1
    if received == 2:
        break
connection.close()

JavaScript:

// Topics: route with word patterns; * matches one word, # matches zero or more.
import amqp from "amqplib";

const conn = await amqp.connect(process.env.AMQP_URL ?? "amqp://localhost");
const ch = await conn.createChannel();
await ch.assertExchange("topic_logs", "topic", { durable: false });
const { queue } = await ch.assertQueue("", { exclusive: true });
await ch.bindQueue(queue, "topic_logs", "*.critical");
for (const key of ["kern.critical", "kern.info", "app.critical"]) ch.publish("topic_logs", key, Buffer.from(key));

let received = 0;
await new Promise((resolve) =>
  ch.consume(queue, (msg) => {
    console.log(` [x] ${msg.content}`);
    if (++received === 2) resolve();
  }, { noAck: true }),
);
await conn.close();

Rust:

//! Topics: route with word patterns; `*` matches one word, `#` matches zero or more.

use firemq_client::{Client, QueueOptions, SubscribeOptions};

#[tokio::main]
async fn main() -> Result<(), firemq_client::Error> {
    let url = std::env::var("FIREMQ_URL").unwrap_or_else(|_| "fire://localhost".into());
    let client = Client::connect(&url).await?;
    client
        .declare_exchange("topic_logs.rust", "topic", Default::default())
        .await?;
    let queue = client.declare_queue("", QueueOptions::temporary()).await?;
    client
        .bind_queue(&queue.name, "topic_logs.rust", "*.critical")
        .await?;

    for key in ["kern.critical", "kern.info", "app.critical"] {
        client.publish("topic_logs.rust", key, key).await?;
    }

    let mut sub = client
        .subscribe(&queue.name, SubscribeOptions::default())
        .await?;
    for _ in 0..2 {
        let Some(msg) = sub.next().await else { break };
        println!(" [x] {}", msg.body_str().unwrap_or_default());
        msg.ack()?;
    }
    client.close().await
}

RPC

A client sends a request with two properties: reply_to names a queue for the answer, and correlation_id matches the answer to the request. The server publishes its answer to reply_to with the same correlation_id.

Python:

"""RPC: a request carries reply_to and correlation_id; the server answers there."""
import os
import threading
import uuid

import pika

URL = pika.URLParameters(os.environ.get("AMQP_URL", "amqp://localhost"))


def server():
    connection = pika.BlockingConnection(URL)
    channel = connection.channel()
    for method, props, body in channel.consume("rpc_queue", inactivity_timeout=5):
        n = int(body)
        channel.basic_publish(
            exchange="",
            routing_key=props.reply_to,
            properties=pika.BasicProperties(correlation_id=props.correlation_id),
            body=str(n * n).encode(),
        )
        channel.basic_ack(method.delivery_tag)
        break
    connection.close()


connection = pika.BlockingConnection(URL)
channel = connection.channel()
channel.queue_declare(queue="rpc_queue")
threading.Thread(target=server).start()

callback_queue = channel.queue_declare(queue="", exclusive=True).method.queue
corr_id = str(uuid.uuid4())
channel.basic_publish(
    exchange="",
    routing_key="rpc_queue",
    properties=pika.BasicProperties(reply_to=callback_queue, correlation_id=corr_id),
    body=b"7",
)
for method, props, body in channel.consume(callback_queue, auto_ack=True, inactivity_timeout=5):
    if props.correlation_id == corr_id:
        print(f" [.] square(7) = {body.decode()}")
        break
connection.close()

Rust:

//! RPC: a request carries `reply_to` and `correlation_id`; the server answers there.

use firemq_client::{Client, PublishOptions, QueueOptions, SubscribeOptions};

#[tokio::main]
async fn main() -> Result<(), firemq_client::Error> {
    let url = std::env::var("FIREMQ_URL").unwrap_or_else(|_| "fire://localhost".into());
    let client = Client::connect(&url).await?;
    client
        .declare_queue("rpc_queue.rust", QueueOptions::default())
        .await?;

    // The server squares numbers.
    let server = client.clone();
    let mut requests = server
        .subscribe("rpc_queue.rust", SubscribeOptions::default())
        .await?;
    tokio::spawn(async move {
        if let Some(req) = requests.next().await {
            let n: u64 = req.body_str().unwrap_or("0").parse().unwrap_or(0);
            let mut reply = PublishOptions::default();
            reply.properties.correlation_id = req.properties.correlation_id.clone();
            let to = req.properties.reply_to.clone().unwrap_or_default();
            let _ = server
                .publish_with("", &to, (n * n).to_string(), reply)
                .await;
            let _ = req.ack();
        }
    });

    // The client sends a request and waits for the matching reply.
    let replies = client.declare_queue("", QueueOptions::temporary()).await?;
    let mut answers = client
        .subscribe(&replies.name, SubscribeOptions::default())
        .await?;
    let mut request = PublishOptions::default();
    request.properties.reply_to = Some(replies.name.clone());
    request.properties.correlation_id = Some("req-1".into());
    client
        .publish_with("", "rpc_queue.rust", "7", request)
        .await?;
    while let Some(answer) = answers.next().await {
        if answer.properties.correlation_id.as_deref() == Some("req-1") {
            println!(" [.] square(7) = {}", answer.body_str().unwrap_or_default());
            answer.ack()?;
            break;
        }
    }
    client.close().await
}

Run all the examples

./examples/run.sh

The script runs every program above against a broker on localhost. It needs uv, Node.js and Rust.