FireMQ

FireMQ is a message broker written in Rust. It speaks RabbitMQ's protocol, AMQP 0-9-1, so the RabbitMQ client you already use connects to FireMQ unchanged. It also speaks its own small protocol for FireMQ's SDKs and for browsers.

FireMQ keeps RabbitMQ's model: publishers send messages to exchanges, exchanges route them to queues through bindings, and consumers acknowledge what they process. It drops what makes RabbitMQ heavy to run: there is no Erlang runtime, no plugins to enable, and no configuration to write before the first message.

firemq

That one command starts a broker with:

• AMQP 0-9-1 on port 5672, for RabbitMQ clients
• the FireMQ protocol on port 5673, for FireMQ SDKs
• the management UI and a RabbitMQ-compatible HTTP API on port 15672
• Prometheus metrics at /metrics and WebSocket at /fire on the same port
• durable storage in ./firemq-data

What you get

• RabbitMQ compatibility. Exchanges (direct, fanout, topic, headers), publisher confirms, mandatory returns, prefetch, TTL, dead-lettering, maximum length, priorities, single active consumer, transactions, policies, vhosts, users and permissions behave as they do in RabbitMQ. CI runs pika, amqplib, amqp091-go, the Java client, the .NET client and rabbitmqadmin against every change.
• Durability you can test. Persistent messages are confirmed only after they are on disk. A test kills the broker with kill -9 while it confirms publishes, restarts it, and checks that no confirmed message is missing.
• A small footprint. One static binary with no runtime dependencies. An idle broker uses about 8 MB of memory on Linux.
• Tools for developers. firemq pub, firemq sub, firemq ctl, firemq bench and firemq doctor work from the terminal, and the broker can run inside Rust tests as a library.

What FireMQ is not (yet)

FireMQ runs on a single node. It does not cluster or replicate, and it does not implement AMQP 1.0, MQTT, STOMP or streams. What FireMQ does not support lists every gap.

Install

FireMQ is a single binary. Choose one of the following methods.

Install script (Linux and macOS)

The script downloads the latest release for your platform, checks its SHA-256 checksum, and installs firemq to /usr/local/bin when that directory is writable, or to ~/.local/bin otherwise.

curl -fsSL https://raw.githubusercontent.com/FireMQ/firemq/main/install.sh | sh

To pin a version or choose the directory, set FIREMQ_VERSION and FIREMQ_INSTALL_DIR:

curl -fsSL https://raw.githubusercontent.com/FireMQ/firemq/main/install.sh | FIREMQ_VERSION=v0.1.0 FIREMQ_INSTALL_DIR=$HOME/bin sh

Docker

docker run -d --name firemq -p 5672:5672 -p 5673:5673 -p 15672:15672 -v firemq-data:/data ghcr.io/firemq/firemq

The image is built FROM scratch and contains only the static binary. Data is stored in the /data volume. Inside the container, the guest user may connect from any host, because connections arrive through the container network rather than from localhost. Set FIREMQ_DEFAULT_USER and FIREMQ_DEFAULT_PASS to create a different first user.

A Compose file is in deploy/docker-compose.yml.

Homebrew (macOS and Linux)

brew install firemq/tap/firemq

To start it at login, with data in Homebrew's var/firemq directory, run brew services start firemq.

GitHub Actions

FireMQ/setup-firemq starts a broker for the rest of a job, on Linux, macOS and Windows runners:

steps:
  - uses: FireMQ/setup-firemq@v1
  - run: npm test   # connect to amqp://guest:guest@localhost:5672

Release archives

Download an archive for Linux (static, x86_64 or arm64), macOS (Apple silicon or Intel) or Windows from the releases page, unpack it, and put firemq on your PATH.

With Cargo

You need Rust 1.89 or later.

cargo install firemq

To build the latest code on main instead, run cargo install --git https://github.com/FireMQ/firemq firemq.

Run as a service on Linux

deploy/firemq.service is a hardened systemd unit that stores data in /var/lib/firemq. Its comments list the commands to install it.

Check the installation

firemq --version
firemq doctor

firemq doctor reports the open-files limit, whether the default ports are free, and whether a broker is running.

Quickstart

This page takes about five minutes. You start a broker, send and receive messages from the terminal and from code, and look at the management UI.

Start the broker

firemq

FireMQ prints its connection strings:

FireMQ 0.1.0 is ready in 40 ms

  AMQP 0-9-1   amqp://guest:guest@localhost:5672
  FireMQ       fire://guest:guest@localhost:5673
  WebSocket    ws://localhost:15672/fire
  Management   http://localhost:15672
  Metrics      http://localhost:15672/metrics
  Data         firemq-data (0 queues, 0 messages recovered)

Leave it running and open a second terminal.

If port 5672 is in use, RabbitMQ is probably running. Stop it, or start FireMQ on other ports, for example firemq --amqp 6672.

Send and receive from the terminal

Declare a queue, publish to it, and read it back:

firemq ctl declare-queue jobs
firemq pub --queue jobs "hello from the terminal"
firemq sub jobs --count 1

firemq pub waits for the broker's confirm, so when it returns the message is on disk. firemq sub prints each message and acknowledges it.

Send and receive from code

Any RabbitMQ client works. With Python and pika:

pip install pika

import pika

connection = pika.BlockingConnection(pika.URLParameters("amqp://guest:guest@localhost:5672"))
channel = connection.channel()
channel.queue_declare(queue="hello")
channel.basic_publish(exchange="", routing_key="hello", body=b"Hello World!")
method, properties, body = channel.basic_get("hello", auto_ack=True)
print(body)
connection.close()

The tutorials cover every messaging pattern in Python, JavaScript and Rust.

Look at the management UI

Open http://localhost:15672 and log in as guest with password guest. The UI shows queues, exchanges, connections and rates, and lets you publish, peek at messages, and manage users and policies.

Next steps

• Read Configuration to change ports, storage and limits.
• Read Migrating from RabbitMQ to move an existing setup.
• Run firemq bench --embedded to measure throughput on your machine.

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.

Clients

You can connect to FireMQ in two ways:

• With a RabbitMQ client over AMQP 0-9-1. Use this to move existing applications, or when your language already has a good AMQP client.
• With a FireMQ SDK over the FireMQ protocol. Use this for new code. The SDKs share one API across languages, confirm every publish by default, and also run in browsers over WebSocket.

Both kinds of client share the same broker: a message published over one protocol can be consumed over the other, with its properties and headers.

RabbitMQ clients

Every AMQP 0-9-1 client works. CI tests these against every change:

Connect with the usual URL, for example amqp://guest:guest@localhost:5672/. See RabbitMQ client compatibility for the details.

FireMQ SDKs

The Python, Ruby, Swift, Dart and C clients are bindings over the Rust SDK, so they ship a native library. Every SDK runs the conformance suite against a real broker in its CI.

Track progress in the roadmap issue.

Connection URLs

fire://user:password@host:5673/vhost      FireMQ protocol over TCP
fires://user:password@host:5674/vhost     over TLS
ws://host:15672/fire?vhost=dev            over WebSocket (browsers)
wss://host/fire                           over secure WebSocket, through a TLS proxy

Missing parts default to user and password guest, vhost / and the standard ports, so fire://localhost is enough for local development.

Rust

[dependencies]
firemq-client = "0.1"
tokio = { version = "1", features = ["full"] }

#![allow(unused)]
fn main() {
use firemq_client::{Client, SubscribeOptions};

async fn run() -> Result<(), firemq_client::Error> {
let client = Client::connect("fire://localhost").await?;
client.declare_queue("jobs", Default::default()).await?;
client.publish("", "jobs", "hello").await?; // returns when the broker confirms

let mut jobs = client.subscribe("jobs", SubscribeOptions::default()).await?;
while let Some(msg) = jobs.next().await {
    println!("{}", msg.body_str().unwrap_or_default());
    msg.ack()?;
}
Ok(())
}
}

The SDK's defaults favour safety: queues and exchanges are durable, messages are persistent, and publish waits for the confirm. Errors are typed: Error::Unroutable for a mandatory message nobody received, Error::Nacked when a queue rejected it, and Error::Server { code, .. } with an AMQP reply code for refused requests.

Reconnection

The Rust client reconnects on its own when the connection drops, for example while the broker restarts. After reconnecting, it does the following:

• Declares again the exchanges, queues and bindings it declared. A queue declared with an empty name gets a new name from the broker, and the client moves its bindings and subscriptions to the new name.
• Resumes every subscription. Your while let Some(msg) = sub.next().await loop keeps running.
• Publishes again any message that was waiting for a confirm. Delivery is at least once, so a consumer can see a message twice.

Messages delivered before the drop and not yet acknowledged go back to the queue, and the broker delivers them again with redelivered set. The subscription skips the old copies, and acknowledging one does nothing.

While the client is reconnecting, calls wait up to 30 seconds by default and then fail with Error::Closed. Attempts back off from 100 ms to 5 s and continue until you call close. To change this, set ConnectOptions::reconnect:

#![allow(unused)]
fn main() {
use std::time::Duration;
use firemq_client::{Client, ConnectOptions, Reconnect};

async fn run() -> Result<(), firemq_client::Error> {
let mut opts = ConnectOptions::from_url("fire://localhost")?;
opts.reconnect = Some(Reconnect {
    max_attempts: Some(10),
    wait: Duration::from_secs(5),
    ..Default::default()
});
let client = Client::connect_with(opts).await?;
Ok(())
}
}

Set reconnect to None to get an error as soon as the connection drops, and handle reconnection yourself.

Write your own client

The FireMQ protocol is small: length-prefixed frames with a MessagePack header and a raw body. A client needs a MessagePack library and a few hundred lines. spec/test-vectors/frames.json has an encoded example of every frame to test your codec against.

Testing with an embedded broker (Rust)

The firemq crate runs a broker inside your tests, on free ports and in memory:

#![allow(unused)]
fn main() {
async fn run() -> std::io::Result<()> {
let server = firemq::builder().ephemeral().random_ports().start().await?;
let amqp_url = server.amqp_url().unwrap();
let fire_url = server.fire_url().unwrap();
// ... connect your code to the broker ...
server.shutdown().await;
Ok(())
}
}

Command-line tool

The firemq binary is the broker and a set of tools for working with it. Run firemq <command> --help for every option.

Manage a broker with ctl

ctl talks to the management API at http://localhost:15672 as guest by default. Change that with --url, --user and --pass, or with FIREMQ_HTTP_URL, FIREMQ_USER and FIREMQ_PASS. --vhost selects the virtual host, and --json prints raw JSON instead of a table.

firemq ctl overview
firemq ctl list queues            # also: exchanges, bindings, connections, consumers, users, vhosts, policies, permissions
firemq ctl declare-queue orders --arg x-message-ttl=60000 --arg x-dead-letter-exchange=dlx
firemq ctl declare-exchange events topic
firemq ctl bind events orders --key 'order.*'
firemq ctl get orders --count 5   # peek; add --take to remove the messages
firemq ctl purge orders
firemq ctl delete-queue orders
firemq ctl add-user app s3cret --tags management
firemq ctl set-permissions app --configure '^app\.' --write '^app\.' --read '^app\.'
firemq ctl set-policy ttl '^tmp\.' '{"message-ttl": 60000}'
firemq ctl export definitions.json
firemq ctl import definitions.json   # also accepts a RabbitMQ export
firemq ctl close-connection '127.0.0.1:50000 -> 127.0.0.1:5672'

Publish and consume

pub and sub use the FireMQ protocol at fire://localhost. Set --url or FIREMQ_URL to change it.

firemq pub events order.created '{"id": 1}' --content-type application/json
firemq pub --queue jobs "one task"           # straight to a queue
firemq pub --queue jobs --count 10000 "load" # publish many, pipelined
cat payload.json | firemq pub --queue jobs   # body from stdin
firemq pub --queue jobs --file payload.bin -H trace=abc

firemq sub jobs                  # print and acknowledge until Ctrl-C
firemq sub jobs --count 10       # stop after 10
firemq sub jobs --peek --count 1 # look without removing (requeue)
firemq sub jobs --json           # JSON lines with properties and headers

Benchmark

firemq bench --embedded                       # an in-process broker, nothing else needed
firemq bench --publishers 4 --consumers 4 --size 1024 --duration 30
firemq bench --persistent                     # durable queue, persistent messages

bench prints published and consumed messages per second every second, then the averages and the end-to-end latency percentiles. Publishers keep --window unconfirmed messages in flight.

Configuration

FireMQ needs no configuration. When you need to change something, each setting can come from three places. The first one found wins:

1. a command-line flag, such as --amqp 6672
2. an environment variable, such as FIREMQ_AMQP=6672
3. the settings file, firemq.toml in the working directory or the file given with --config

firemq init writes a firemq.toml that lists every setting with its default and a comment. FireMQ rejects unknown keys in the file, so a typo fails at startup instead of being ignored.

Listeners

A listener value is host:port, a bare port (all interfaces), or off.

In firemq.toml, listeners go in [listeners] and the TLS files in [tls] as cert and key.

Storage

With fsync = false, a confirmed persistent message survives a broker crash but can be lost if the machine loses power. See Durability and recovery.

Limits

Sizes take units: 512MB, 2GiB, 1000000. When a limit is crossed, publishers are paused; see Operations.

Users

The first user is created only when the broker has no users, and gets the administrator tag and full access to the / vhost. An empty FIREMQ_LOOPBACK_USERS lets every user connect from any host.

Logging

The level also accepts per-module filters, for example info,firemq_amqp=debug.

Development mode

firemq --dev is shorthand for an in-memory broker that accepts guest from any host and logs at debug level. Use it in containers for local development and in CI.

Management UI, API and metrics

FireMQ's HTTP listener, port 15672 by default, serves four things:

UI

Open http://localhost:15672 and log in with a user that has the administrator, management, monitoring or policymaker tag. The UI shows totals and rates, connections, queues and exchanges. From a queue's page you can publish, peek at or take messages, bind it, purge it and delete it. The Admin page manages users, vhosts, policies and definitions.

API

The API follows RabbitMQ's management API: the same paths, JSON shapes and status codes. Tools written for RabbitMQ, including rabbitmqadmin, work with it. Authenticate with HTTP Basic:

curl -u guest:guest http://localhost:15672/api/overview
curl -u guest:guest http://localhost:15672/api/queues/%2F
curl -u guest:guest -X PUT -H 'content-type: application/json' \
  -d '{"durable": true, "arguments": {"x-max-length": 1000}}' \
  http://localhost:15672/api/queues/%2F/orders

The default vhost / is written %2F in URLs, and the default exchange is amq.default.

Access

A user without one of these tags cannot use the API. Users in loopback_users can only log in from localhost.

Endpoints

List endpoints accept ?columns=name,messages (dotted paths such as message_stats.publish work) and ?name= to filter by name.

Prometheus metrics

/metrics needs no authentication. Metric names follow RabbitMQ's Prometheus plugin, so existing dashboards work:

• rabbitmq_connections, rabbitmq_channels, rabbitmq_consumers, rabbitmq_queues
• rabbitmq_queue_messages_ready, rabbitmq_queue_messages_unacked, rabbitmq_queue_messages
• rabbitmq_global_messages_received_total, _delivered_total, _acknowledged_total, _confirmed_total, _unroutable_dropped_total
• rabbitmq_process_resident_memory_bytes, rabbitmq_resident_memory_limit_bytes, rabbitmq_disk_space_available_bytes, rabbitmq_alarms_memory_used_watermark, rabbitmq_alarms_free_disk_space_watermark

Per-queue series carry vhost and queue labels: firemq_queue_messages_ready, firemq_queue_messages_unacked, firemq_queue_consumers, firemq_queue_messages_published_total, firemq_queue_messages_delivered_total and firemq_queue_messages_acked_total.

# prometheus.yml
scrape_configs:
  - job_name: firemq
    static_configs:
      - targets: ["localhost:15672"]

Durability and recovery

This page explains what survives a restart or a crash, and what a publisher confirm promises.

What survives a restart

A message survives a restart when both of these are true:

• the queue is durable (durable: true when declared)
• the message is persistent (delivery_mode: 2 in AMQP, persistent: true in the FireMQ protocol)

Durable exchanges, durable queues, the bindings between them, users, permissions, vhosts and policies also survive. Transient messages and non-durable queues are kept in memory and are gone after a restart, as in RabbitMQ. Messages that were delivered but not acknowledged come back after a restart with redelivered set.

--ephemeral keeps everything in memory, including durable queues.

What a confirm means

With publisher confirms (confirm.select in AMQP, always on in the FireMQ protocol), the broker acknowledges a publish when it has taken responsibility for the message:

Syncing is batched. One thread waits about 2 ms after the first pending write, then syncs every queue file that changed and confirms everything written before the sync. Under load, one sync covers thousands of publishes.

With --fsync=false, confirms are sent once the message is written to the operating system. The message survives a broker crash but not a power loss.

How FireMQ stores messages

firemq-data/
  node.json              format version
  definitions.json       users, vhosts, permissions, policies, durable topology
  firemq.lock            prevents two brokers from using the directory
  queues/<vhost>/<queue>/
    meta.json            queue name, vhost and arguments
    0000000001.seg       messages, appended in order
    0000000001.ack       which messages were acknowledged

• Each durable queue appends messages to segment files of about 32 MiB. Every record carries a CRC32C checksum.
• When a message is acknowledged, its position is appended to the segment's .ack file. A segment is deleted as soon as all its messages are acknowledged, so disk use follows the backlog.
• A durable queue keeps up to 8 MiB of message bodies in memory and reads the rest from disk, so a large backlog does not fill memory.
• definitions.json is rewritten atomically (write, sync, rename) shortly after any durable change. It uses RabbitMQ's definitions format, so you can read it, back it up, or import it into RabbitMQ.

Crash recovery

At startup FireMQ:

1. reads definitions.json and recreates users, vhosts, policies, exchanges, queues and bindings
2. restores queues that have data on disk but are missing from the definitions, which happens if the broker crashed right after a queue was declared
3. scans each queue's segments, verifies every checksum, and truncates a segment at the first damaged record, which is where a crash interrupted a write
4. skips acknowledged and transient messages and requeues the rest as redelivered

The startup banner and the Data line report how many queues and messages were recovered.

Tested behaviour

A test in the repository (crates/firemq/tests/chaos.rs) starts the broker, publishes persistent messages with 256 confirms in flight, kills the process with kill -9 at a random moment, restarts it, and repeats four times. It then checks that every confirmed message is present. Tens of thousands of messages are confirmed per run, and none are lost.

Security

Users and passwords

On first start, FireMQ creates one user, guest with password guest by default, with the administrator tag and full access to the / vhost. Like RabbitMQ, guest can only connect from localhost. Before exposing FireMQ to a network:

1. Create your own users and grant them access. These commands run as guest, from the broker's machine:

   firemq ctl add-user admin 'a-long-password' --tags administrator
   firemq ctl set-permissions admin
   firemq ctl add-user orders-service 'another-password'
   firemq ctl set-permissions orders-service --configure '^orders\.' --write '^orders\.' --read '^orders\.'
   

2. Delete guest, logged in as the new administrator:

   firemq ctl delete-user guest --user admin --pass 'a-long-password'
   

Alternatively, set FIREMQ_DEFAULT_USER and FIREMQ_DEFAULT_PASS before the first start, so guest is never created.

Passwords are stored as salted SHA-256 hashes in RabbitMQ's format, so users exported from RabbitMQ keep their passwords when imported.

Permissions

Permissions work as in RabbitMQ. Each user has, per vhost, three regular expressions matched against queue and exchange names:

An empty pattern grants nothing. Publishing to the default exchange checks write access to amq.default. A user without permissions on a vhost cannot connect to it.

TLS

Pass a PEM certificate chain and private key:

firemq --tls-cert /etc/firemq/cert.pem --tls-key /etc/firemq/key.pem

This starts AMQPS on 5671 and the FireMQ protocol over TLS on 5674, next to the plain listeners. To accept only TLS, turn the plain ones off with --amqp off --fire off. FireMQ uses rustls and supports TLS 1.2 and 1.3.

The HTTP listener serves plain HTTP. To serve the UI, API and WebSocket over HTTPS, put a TLS-terminating proxy such as Caddy, nginx or a cloud load balancer in front of port 15672.

Network exposure

Every listener binds to all interfaces by default. To keep a port local, bind it to the loopback address, for example --http 127.0.0.1:15672.

Operations

Health checks

Resource alarms

FireMQ samples its memory use and the free disk space of the data directory every second. It raises an alarm when:

• resident memory exceeds the memory high watermark: 40% of system memory by default, or the absolute --memory-limit
• free disk space drops below --disk-free-limit, 50 MB by default

While an alarm is active, FireMQ stops reading from connections that publish. Consumers keep working, so queues drain and memory is freed. AMQP clients that support it receive connection.blocked and connection.unblocked. FireMQ SDKs receive a flow frame. The UI shows a banner, and rabbitmq_alarms_* metrics are 1.

Monitoring

Scrape /metrics with Prometheus; see Management UI, API and metrics. Useful alerts:

• rabbitmq_alarms_memory_used_watermark == 1 or rabbitmq_alarms_free_disk_space_watermark == 1: publishers are paused
• firemq_queue_messages_ready growing steadily: consumers are not keeping up
• firemq_queue_consumers == 0 on a queue that should be consumed

Logs

Logs go to standard error. Use --log-format json for log collectors, and --log-level debug or a filter such as info,firemq_amqp=debug to investigate client behaviour.

Backups

• Topology and users: firemq ctl export definitions.json, or copy firemq-data/definitions.json. Restore with firemq ctl import.
• Messages: stop the broker and copy the whole data directory. Copying it while the broker runs can capture a segment in the middle of a write; recovery truncates such a segment, losing the messages after the damaged record.

Upgrades

Stop the broker, replace the binary, and start it again. FireMQ refuses to start on a data directory written by a newer format than it understands, rather than misreading it.

Open files

Each connection and each segment file uses a file descriptor. FireMQ raises its soft limit to the hard limit at startup; firemq doctor shows both. For many connections or large backlogs, set the hard limit to at least 65536 (LimitNOFILE in the systemd unit).

Migrating from RabbitMQ

Most applications move to FireMQ without code changes, because FireMQ speaks AMQP 0-9-1 and serves RabbitMQ's management API.

1. Check that you use supported features

Read What FireMQ does not support. The main questions:

• Do you rely on clustering, quorum queue replication, streams, federation or the shovel plugin?
• Do you use plugin exchange types, such as x-delayed-message or x-consistent-hash?
• Do you connect with MQTT, STOMP or AMQP 1.0?

If not, your applications should work unchanged.

2. Export your definitions from RabbitMQ

rabbitmqadmin export rabbitmq-definitions.json
# or: curl -u admin:password http://rabbitmq:15672/api/definitions > rabbitmq-definitions.json

3. Import them into FireMQ

firemq ctl import rabbitmq-definitions.json

Users keep their passwords, because FireMQ verifies RabbitMQ's SHA-256 and SHA-512 password hashes. Vhosts, permissions, policies, exchanges, queues and bindings are created. Queues declared as x-queue-type: quorum become durable FireMQ queues.

4. Point your applications at FireMQ

Change the host in your AMQP URLs. Ports, vhosts and credentials stay the same.

5. Move the messages, if you need to

Definitions do not contain messages. To move messages that are still queued, either let the RabbitMQ queues drain before switching publishers, or run a small program that consumes from RabbitMQ and publishes to FireMQ with confirms.

Differences you may notice

• Management UI. FireMQ's UI is different from RabbitMQ's and smaller. The API underneath is compatible.
• Product name. connection.start and /api/overview report the product as FireMQ. Code that checks for RabbitMQ by name, which is rare, needs a change.
• Message TTL. Expired messages are removed when they reach the head of the queue, as in RabbitMQ classic queues, with a timer precision of a few milliseconds.
• Storage. Durable queues always page to disk instead of switching between lazy and default modes; x-queue-mode is accepted and ignored.
• Errors. Reply codes and messages follow RabbitMQ's, including PRECONDITION_FAILED - inequivalent arg ..., so existing error handling keeps working.

RabbitMQ client compatibility

FireMQ speaks AMQP 0-9-1, so RabbitMQ client libraries connect to it unchanged. CI runs every client below against the firemq binary on every pull request.

The suites live in compat/. To run one locally, start firemq and set AMQP_URL:

AMQP_URL=amqp://guest:guest@localhost:5672/ uv run --with pika --with pytest pytest compat/python

Behaviour that matches RabbitMQ on purpose

• basic.publish with immediate=true closes the connection with 540 NOT_IMPLEMENTED.
• channel.flow with active=false is answered with 540 NOT_IMPLEMENTED.
• basic.qos with a non-zero prefetch_size is answered with 540 NOT_IMPLEMENTED.
• Deleting a queue or exchange that does not exist succeeds.
• The guest user can only connect from localhost. Set FIREMQ_LOOPBACK_USERS= to lift the restriction, for example in containers.

Management API

FireMQ serves a subset of RabbitMQ's management HTTP API on port 15672 with the same paths, JSON shapes and status codes. CI drives it with RabbitMQ's own rabbitmqadmin script: show overview, declare and delete exchanges, queues, bindings and policies, publish, get, list and export definitions.

Covered endpoints: /api/overview, /api/nodes, /api/whoami, /api/vhosts, /api/users, /api/permissions, /api/policies, /api/exchanges (including /publish and bindings), /api/queues (including /get, /contents and bindings), /api/bindings, /api/connections, /api/channels, /api/consumers, /api/definitions (export and import, whole broker or one vhost), /api/aliveness-test/{vhost} and /api/health/checks/*.

Prometheus metrics are at /metrics on the same port, using RabbitMQ's metric names (rabbitmq_queue_messages_ready, rabbitmq_global_messages_received_total and so on) plus per-queue firemq_queue_* series. /healthz is an unauthenticated liveness probe.

What FireMQ does not support

FireMQ aims to be honest about its gaps. If something you need is missing, open an issue on GitHub.

Not supported

Supported

Everything else in RabbitMQ's core AMQP 0-9-1 model works, including:

• exchanges: direct, fanout, topic and headers (x-match all, any, all-with-x and any-with-x), exchange-to-exchange bindings, alternate exchanges, internal and auto-delete exchanges
• queues: durable, exclusive, auto-delete, server-named, x-message-ttl, per-message expiration, x-expires, x-max-length, x-max-length-bytes, x-overflow (drop-head, reject-publish, reject-publish-dlx), x-dead-letter-exchange, x-dead-letter-routing-key with x-death headers, x-max-priority, x-single-active-consumer
• consuming: prefetch per consumer and per channel (global), acks, nacks and rejects with requeue, basic.get, basic.recover, consumer cancel notifications
• publishing: publisher confirms, mandatory returns, transactions (tx.select, tx.commit, tx.rollback)
• policies, vhosts, users, permissions, definitions import and export, memory and disk alarms with connection.blocked

The FireMQ protocol

Version 1, release candidate. The version freezes once the Rust, Go and TypeScript SDKs pass the conformance suite; from then on, changes are additive only and gated by capabilities.

The FireMQ protocol is the native protocol of the FireMQ broker. It carries the same model as AMQP 0-9-1 (exchanges, queues, bindings, acknowledgements, publisher confirms) in a format that is small enough to implement in an afternoon in any language with a MessagePack library.

FireMQ also speaks AMQP 0-9-1 on port 5672. Both protocols share one broker, so a message published over one is consumed over the other.

Design rules

• One connection is one session. There are no channels.
• Every frame has a MessagePack header and an optional raw payload. Message bodies are never MessagePack-encoded.
• Requests carry a request id (rid) and get exactly one reply: ok, error, or an op-specific reply.
• Omitted header fields take their zero value: false, 0, empty string, empty map. Implementations ignore header keys they do not know.
• All integers in framing are big-endian.

Transports

Connection URLs use the schemes fire://, fires:// (TLS), ws:// and wss://:

fire://user:password@host:5673/vhost

For fire and fires, the vhost is the URL path with the leading / removed and percent-decoded; an empty path means vhost /. For ws and wss, the path is the WebSocket endpoint (/fire when empty) and the vhost is the vhost query parameter, for example ws://localhost:15672/fire?vhost=dev. The default user and password are guest and guest. The default ports are 5673 (fire), 5674 (fires), 15672 (ws) and 15671 (wss).

Frame format

On TCP and TLS:

+-----------+--------+------------+---------------------+------------------+
| len : u32 | op : u8| hlen : u32 | header (hlen bytes) | payload          |
+-----------+--------+------------+---------------------+------------------+

• len counts every byte after the len field: 1 + 4 + hlen + payload length.
• header is a MessagePack map with string keys. hlen may be 0, meaning an empty map.
• payload is the rest of the frame. Only publish, return, deliver and get_ok carry a payload.

On WebSocket the frame is the same without len.

Before welcome, a frame may be at most 65 536 bytes. After welcome, the limit is welcome.max_frame (16 MiB by default). A message body must fit in one frame.

Op codes

A server that receives an unknown op replies error with code 540 and keeps the connection open.

Connection lifecycle

client                                   server
  | -- FMQ1 (TCP only) ------------------> |
  | -- hello ----------------------------> |
  | <-------------------- welcome or error |
  | ... requests, publishes, deliveries ...|
  | -- close ----------------------------> |
  | <------------------------------- close |

hello

welcome

If authentication fails, the server replies error with code 403 and fatal: true, then closes the socket.

Heartbeats

When the negotiated interval h is non-zero, each side sends ping when it has sent nothing for h / 2 seconds. Each side considers the peer dead when it has received nothing for 2 × h seconds. A peer that receives ping replies pong.

close

{"code": uint, "msg": str}. The receiver replies with its own close and closes the socket. Code 200 means a normal shutdown.

Replies

ok

error

flow

{"pause": bool}. The server sends pause: true when a memory or disk alarm fires. Clients stop publishing until pause: false. A client that keeps publishing is throttled by the server no longer reading its socket.

Topology

declare_exchange

Reply: ok. Redeclaring an existing exchange with different type, durable, auto_delete or internal fails with 406.

delete_exchange

{"rid", "name", "if_unused": bool}. Reply: ok.

declare_queue

Reply: ok with name, messages and consumers.

delete_queue

{"rid", "name", "if_unused": bool, "if_empty": bool}. Reply: ok with messages, the number of messages deleted.

purge_queue

{"rid", "name"}. Reply: ok with messages, the number of messages purged.

bind and unbind

Reply: ok.

Publishing

publish

The payload is the message body.

Message properties:

confirm

The server confirms every publish with a non-zero seq exactly once.

• {"upto": n} confirms every outstanding publish with seq <= n. The broker has taken responsibility for those messages: they are on disk when persistent and routed to a durable queue, and in memory otherwise.
• {"seq": n, "nack": true, "code": uint, "msg": str} rejects one publish, for example because a queue with x-overflow: reject-publish is full. The message was not enqueued in every queue that should have received it.

A nack for a sequence number is always sent before any upto that covers it.

return

{"seq", "code", "msg", "exchange", "key", "props", "headers"} with the message body as payload. Sent for a mandatory publish that no queue received, with code 312 (no-route). The return is sent before the publish's confirm.

Consuming

subscribe

Reply: ok.

credit

{"sub": uint, "credit": uint} adds credit to a subscription. The server sends at most as many deliver frames as the credit granted so far. Clients usually grant their buffer size at subscribe, then top up by half that size each time they have processed half of it.

deliver

The payload is the message body.

ack and nack

• ack: {"tag": uint, "multiple": bool}. With multiple, acknowledges every unacknowledged delivery with a tag up to and including tag.
• nack: {"tag", "multiple", "requeue": bool}. Requeued messages are delivered again with redelivered: true. Messages that are not requeued are dead-lettered when the queue has a dead-letter exchange, and dropped otherwise.

Neither has a reply. An unknown tag produces a non-fatal error with code 406.

unsubscribe

{"rid", "sub"}. Reply: ok. The server sends no more deliveries for sub. Unacknowledged deliveries stay unacknowledged until the client acks them or the connection closes, when they are requeued.

cancelled

{"sub", "code", "msg"}. The server ended a subscription, for example because its queue was deleted.

get and get_ok

get is {"rid", "queue", "no_ack": bool}. When a message is ready, the server replies get_ok:

When the queue is empty, the reply is ok with empty: true.

Interoperability with AMQP 0-9-1

Reserved for later versions

These names are reserved so later versions can add them without breaking version 1 clients:

• Streams: declare_queue with args["x-queue-type"] = "stream", a subscribe field offset ("first", "last", "next", a uint, or {"ts": uint}), a deliver field offset, and an op 0x39 commit_offset.
• Capabilities streams, zstd and scram.

Test vectors

test-vectors/frames.json lists one encoded frame per op with its decoded header and payload. Decoders must decode every vector to the listed header. Encoders do not have to reproduce the bytes exactly, because MessagePack allows several encodings of the same map.

SDK conformance suite

Every FireMQ SDK runs the same conformance scenarios. The runner, firemq-conformance, starts an in-memory broker and drives the SDK through a small driver: a program, written with the SDK, that reads JSON commands on stdin and writes JSON replies on stdout. A driver is usually 150 to 300 lines.

cargo run -p firemq-conformance -- run --driver "go run ./conformance/driver"

The runner prints one line per scenario and exits with status 1 if any scenario fails. --only <name> runs one scenario and --list lists them.

Driver protocol

The runner writes one JSON object per line. Each has an id, an op, and the op's fields. The driver writes exactly one reply line per command, with the same id, in order:

{"id": 3, "op": "publish", "exchange": "", "key": "jobs", "body": "hi", "persistent": true, "mandatory": false, "properties": {}, "headers": {}}
{"id": 3, "ok": true, "result": {}}

A failed command replies with ok: false and an error:

{"id": 4, "ok": false, "error": {"kind": "server", "code": 406, "message": "PRECONDITION_FAILED - ..."}}

The driver must not write anything else to stdout. It may log to stderr. It exits when stdin closes.

Ops

The runner always sends every field listed, so drivers need no defaults.

args and headers are JSON objects whose values are strings, numbers, booleans, nulls, arrays or objects. properties uses the keys of the protocol's message properties: content_type, content_encoding, correlation_id, reply_to, expiration (milliseconds), message_id, timestamp, type, user_id, app_id and priority.

A message is:

{
  "body": "text, or null when the body is not UTF-8",
  "body_base64": "the body, base64-encoded",
  "exchange": "orders",
  "routing_key": "order.created",
  "redelivered": false,
  "tag": "17",
  "properties": {"content_type": "application/json"},
  "headers": {"trace": "abc"}
}

tag is an opaque string the runner passes back to ack and nack.

Scenarios

Benchmarks

This page compares FireMQ with RabbitMQ on the same machine, under the same limits, using the same load generator. Run bench/run.sh to reproduce it on your hardware; numbers from a laptop are a comparison, not a capacity plan.

Method

Each broker runs in a Docker container limited to 4 CPUs and 2 GiB of memory. The load generators run in a second container, also limited to 4 CPUs, on the same Docker network. Nothing crosses the macOS host's network stack.

Two load generators produce the same workload:

• amqp_bench, an AMQP 0-9-1 client built on amqprs, drives both brokers. Its source is crates/firemq-amqp/examples/amqp_bench.rs.
• firemq bench drives FireMQ over its own protocol.

Both work the same way:

• Every publisher uses publisher confirms with up to 1,000 unconfirmed messages in flight.
• Every consumer uses a prefetch of 1,000 and acknowledges each message individually.
• The queue is durable. "Persistent" scenarios publish persistent messages, which both brokers write to disk before confirming; "transient" scenarios publish non-persistent messages.
• Each message body starts with its send time, so the consumer measures end-to-end latency from publish to delivery.
• Throughput scenarios publish as fast as confirms allow. Latency scenarios publish a fixed 1,000 messages per second, so latency isn't dominated by queueing.

A fresh broker container starts for every run. Each scenario runs three times for 20 seconds, and the tables show the median run: the run with the median delivered rate. Memory is the container's peak memory use during the run, sampled every 2 seconds with docker stats.

Throughput is the rate at which consumers received messages. When a broker accepts messages faster than it delivers them, the queue grows, and latency then measures mostly time spent waiting in the queue. The results call out those runs.

The recovery test fills a durable queue with 2,000,000 persistent 256-byte messages, stops the broker gracefully, starts it again, and measures how long until the queue holds every message again. It counts messages with a passive queue.declare, which both brokers answer exactly; RabbitMQ's management API refreshes its counts only every 5 seconds. Startup and recovery times include about 0.2 seconds to start the container that measures them.

Results

Measured on November 7, 2026, on an Apple M2 Pro with 10 cores and Docker Desktop 29.4.3. FireMQ ran at commit bf67ac7. RabbitMQ ran from the rabbitmq:4-management image: RabbitMQ 4.4.0 on Erlang/OTP 29. RabbitMQ's numbers come from a run earlier the same evening. FireMQ's runs were repeated after a change that lowered its idle memory, which doesn't affect RabbitMQ. The raw logs and CSV files are in bench/results/2026-11-07.

Throughput

Messages delivered to consumers per second:

Driven by the same AMQP client, FireMQ delivered 1.2 to 4.6 times as many messages per second as RabbitMQ. The gap is widest with persistent messages and several publishers, because FireMQ syncs the writes of all publishers together, at most once every 2 ms. The FireMQ protocol adds another 10 to 25 percent.

In the two AMQP scenarios with 4 publishers, FireMQ accepted messages faster than it delivered them:

• Transient messages: 313,000 published and 274,000 delivered per second.
• Persistent messages: 283,000 published and 261,000 delivered per second.

The broker was at its 4-CPU limit in both. FireMQ doesn't slow publishers down when consumers fall behind, so the queue grew for the length of the run. RabbitMQ slows publishers in that situation with its credit flow. Over the FireMQ protocol, whose acknowledgments are cumulative and cheaper to process, publishing and delivery stayed level in every scenario.

Latency

End-to-end latency with publishers held at 1,000 messages per second, p50 / p99:

Persistent messages include the wait for the disk: neither broker confirms or delivers them before the write that holds them is synced.

End-to-end latency at full throughput, p50 / p99:

At full throughput, latency mostly measures time spent waiting. When the consumer is the bottleneck, it holds up to its prefetch of 1,000 messages that it hasn't processed yet, so latency approaches 1,000 divided by the delivery rate: about 7 ms at 140,000 messages per second. The FireMQ AMQP rows with 4 publishers show the queue that grew during those runs.

Memory

Peak broker memory during the throughput runs:

Most of RabbitMQ's memory is the Erlang runtime, which uses about 245 MiB before the first message. FireMQ's memory follows the messages it holds. In the AMQP runs with 4 publishers, that includes the backlog that built up during the run.

Startup and recovery

On startup FireMQ reads each queue's segment files and verifies every record's checksum, so recovery time grows with the amount of queued data on disk. See Durability and recovery.

Reproduce

bench/run.sh          # 20 seconds per run, 3 runs
bench/run.sh 60 5     # longer runs
CPUS=2 MEMORY=1g bench/run.sh

The script builds the FireMQ image from the working tree, builds the load generators for Linux, and writes raw logs and CSV files to bench/results/<date>/. To turn them into the tables on this page, run:

bench/summarize.py bench/results/<date>