AMQP — the RabbitMQ protocol

OxiDB speaks AMQP 0-9-1, the protocol RabbitMQ clients use. Code written for RabbitMQ — pika (Python), amqplib (Node), RabbitMQ.Client (.NET), amqp091-go (Go), the official tutorials included — points at OxiDB and works unmodified. It adds the one semantic MQTT cannot express: a work queue, where each message goes to exactly one of a pool of competing consumers. Off by default.

Enable it

# start the AMQP listener (5672 is the AMQP default port)
OXIDB_AMQP_PORT=5672 ./oxidb-server

Hello world — RabbitMQ's own tutorial code, unmodified

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1', 5672))
ch = conn.channel()
ch.queue_declare(queue='hello')
ch.basic_publish(exchange='', routing_key='hello', body=b'Hello OxiDB!')

method, props, body = ch.basic_get('hello', auto_ack=True)
print(body)          # b'Hello OxiDB!'
conn.close()

Work queues — competing consumers

An MQTT subscription copies every message to every subscriber. An AMQP queue hands each message to exactly one of its consumers, round-robin — work distribution across a worker pool. Unacked messages requeue (flagged redelivered) if a worker dies.

# worker.py — run as many of these as you like; they split the work
def on_message(ch, method, props, body):
    do_work(body)
    ch.basic_ack(method.delivery_tag)   # ack AFTER the work: crash = requeue

ch.basic_qos(prefetch_count=1)          # a busy worker's turn passes to an idle one
ch.basic_consume('tasks', on_message)
ch.start_consuming()

Exchanges

ch.exchange_declare(exchange='logs', exchange_type='topic')
ch.queue_bind('kernel-q', 'logs', 'kern.*')
ch.queue_bind('audit-q',  'logs', '#')
ch.basic_publish('logs', 'kern.crit', b'disk on fire')   # reaches both

Publisher confirms & mandatory

In confirm mode the broker acks each publish — and for a persistent message the ack is only sent after the fsync (write-before-confirm). An unroutable mandatory publish comes back as Basic.Return (312 NO_ROUTE) instead of vanishing; pika surfaces it as UnroutableError.

// .NET — RabbitMQ.Client 7.x, unmodified
var factory = new ConnectionFactory { HostName = "127.0.0.1", Port = 5672 };
await using var conn = await factory.CreateConnectionAsync();
await using var ch = await conn.CreateChannelAsync(new CreateChannelOptions(
    publisherConfirmationsEnabled: true,
    publisherConfirmationTrackingEnabled: true));

await ch.QueueDeclareAsync("orders", durable: true, exclusive: false, autoDelete: false);
await ch.BasicPublishAsync("", "orders", mandatory: false,
    basicProperties: new BasicProperties { Persistent = true },
    body: JsonSerializer.SerializeToUtf8Bytes(order));
// awaited = the broker has fsync'd it. A crash after this point cannot lose it.

Durability follows the protocol, not a config flag

AMQP already says what should survive: a queue declared durable holding messages published with delivery_mode=2 (persistent) is written through the document engine's WAL and survives a SIGKILL — recovered messages arrive flagged redelivered, acknowledged ones stay consumed. Everything else lives in memory, which is what the client asked for by not saying durable. There is no OXIDB_AMQP_PERSIST because the protocol makes the choice per queue and per message.

Under the hood, pipelined persistent publishes are batched into one fsync per burst, and bursts from different connections share fsync rounds through a group committer — see the numbers below.

MQTT ↔ AMQP bridge

The pre-declared amq.topic exchange bridges the two brokers the same way RabbitMQ's own MQTT plugin does: MQTT topic slashes become AMQP routing-key dots and back, MQTT QoS ≥ 1 maps to persistent. A sensor publishes MQTT; a worker pool consumes AMQP — one binary.

# both listeners on
OXIDB_MQTT_PORT=1883 OXIDB_AMQP_PORT=5672 ./oxidb-server

# AMQP side: bind a queue to amq.topic (pre-declared, no declare needed)
ch.queue_declare(queue='readings', durable=True)
ch.queue_bind('readings', 'amq.topic', 'sensors.#')

# MQTT side: publish normally…
mosquitto_pub -p 1883 -t sensors/floor1/temp -m "21.5"
# …the AMQP consumer receives it with routing key sensors.floor1.temp

The reverse works too: an AMQP publish to amq.topic with routing key alerts.fire reaches MQTT subscribers of alerts/+ (and OxiMem RESP subscribers on the same bus).

Authentication

OXIDB_AMQP_PORT=5672 OXIDB_AMQP_USER=app OXIDB_AMQP_PASSWORD=secret ./oxidb-server
# clients use PLAIN auth exactly as against RabbitMQ: amqp://app:secret@host:5672/

Performance vs RabbitMQ

Measured with the same Go client (amqp091-go), the same code path, against RabbitMQ 4.x on the same machine (100-byte bodies; harness in tests/rabbitmq-benchmark-go):

The one loss is deliberate: every OxiDB durable confirm sits behind a real F_FULLFSYNC of its batch, while RabbitMQ classic queues flush lazily on an interval — its persistent confirm does not, by itself, prove the message is on stable storage. Add concurrency and the honest fsync wins anyway.

The subset, stated honestly

Implemented: connections/channels/heartbeats, PLAIN auth, default + direct/fanout/topic exchanges, queue declare/bind (durable, exclusive, auto-delete, server-named, passive), publish with multi-frame bodies, consume/get/ack/nack/reject, Basic.Qos prefetch, publisher confirms, mandatory Basic.Return, and the MQTT bridge.

Not implemented — and refused with a clear channel error, never silently accepted: the tx class, headers exchanges, exchange-to-exchange bindings, per-message TTL/priority, dead-letter arguments, alternate exchanges, streams, and AMQP 1.0 (a different protocol entirely).

AMQP queues are node-local, like the MQTT broker: consumers are connections and connections live on one node; the durable data behind a queue is what replicates in cluster mode.