AMQP 0-9-1

RabbitMQ's protocol, for pika, amqplib, amqp091-go, the RabbitMQ Java and .NET clients and every other AMQP 0-9-1 library.

Endpoint and credentials

Note the port: 5673, not AMQP's usual 5671, which the Service Bus endpoint on the same host needs. The certificate is a public Let's Encrypt one for amqp.powerqueue.link, so the default TLS settings of each library verify it. Put the port in the URL; most libraries otherwise assume 5671 for amqps.

Python (pika)

pip install pika. A publisher with publisher confirms, so a refusal (a full queue, a message outliving the lifetime limit) raises instead of passing silently:

import pika

URL = "amqps://user:password@amqp.powerqueue.link:5673/<your-identifier>"

with pika.BlockingConnection(pika.URLParameters(URL)) as connection:
    channel = connection.channel()
    channel.queue_declare("tasks", durable=True)
    channel.confirm_delivery()
    channel.basic_publish(
        exchange="",
        routing_key="tasks",
        body=b'{"id": 1}',
        properties=pika.BasicProperties(content_type="application/json"),
    )
    print("published and confirmed")

A consumer:

import pika

URL = "amqps://user:password@amqp.powerqueue.link:5673/<your-identifier>"


def handle(channel, method, properties, body):
    print("received", body)
    channel.basic_ack(method.delivery_tag)


with pika.BlockingConnection(pika.URLParameters(URL)) as connection:
    channel = connection.channel()
    channel.basic_qos(prefetch_count=10)
    channel.basic_consume("tasks", handle)
    channel.start_consuming()

Node.js (amqplib)

npm install amqplib, then run as an ES module (send.mjs):

import amqp from "amqplib";

const url = "amqps://user:password@amqp.powerqueue.link:5673/<your-identifier>";
const connection = await amqp.connect(url);
const channel = await connection.createConfirmChannel();
await channel.assertQueue("tasks", { durable: true });

channel.sendToQueue("tasks", Buffer.from(JSON.stringify({ id: 1 })), { contentType: "application/json" });
await channel.waitForConfirms();

await channel.consume("tasks", (message) => {
  console.log("received", message.content.toString());
  channel.ack(message);
});
setTimeout(() => connection.close(), 2000);

Go (amqp091-go)

go get github.com/rabbitmq/amqp091-go:

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	connection, err := amqp.Dial("amqps://user:password@amqp.powerqueue.link:5673/<your-identifier>")
	if err != nil {
		log.Fatal(err)
	}
	defer connection.Close()
	channel, err := connection.Channel()
	if err != nil {
		log.Fatal(err)
	}
	if _, err := channel.QueueDeclare("tasks", true, false, false, false, nil); err != nil {
		log.Fatal(err)
	}
	if err := channel.Confirm(false); err != nil {
		log.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	confirmation, err := channel.PublishWithDeferredConfirmWithContext(ctx, "", "tasks", false, false,
		amqp.Publishing{ContentType: "text/plain", Body: []byte("hello")})
	if err != nil {
		log.Fatal(err)
	}
	if acked, err := confirmation.WaitContext(ctx); err != nil || !acked {
		log.Fatalf("not confirmed: %v", err)
	}
	message, ok, err := channel.Get("tasks", true)
	if err != nil {
		log.Fatal(err)
	}
	if ok {
		fmt.Println("received", string(message.Body))
	}
}

What is supported

• Exchanges of type direct, fanout and topic, and the default exchange; exchange.declare and exchange.delete; queue.declare, bind, unbind, purge and delete.
• basic.publish (with mandatory), consume, cancel, get, ack, nack, reject, qos (per consumer and global), recover.
• Publisher confirms (confirm.select), consumer cancel notifications and basic.nack.
• Queue arguments: x-message-ttl, x-expires, x-max-length, x-max-length-bytes, x-overflow, x-dead-letter-exchange, x-dead-letter-routing-key, x-delivery-limit and x-consumer-timeout. Per-message expiration. Dead-lettered messages carry x-death headers.
• Negotiated limits: up to 64 channels per connection, frames up to 131,072 bytes, messages up to 256 KiB (advertised as max_message_size in the server properties).

Differences from RabbitMQ

• No headers exchange (declaring one is a channel error, COMMAND_INVALID) and no exchange-to-exchange bindings.
• No transactions: the tx class answers NOT_IMPLEMENTED. Use publisher confirms.
• x-queue-type takes PowerQueue's own values (fifo, priority, sorted, random), not RabbitMQ's classic, quorum or stream; leave it out for an ordinary queue.
• Only SASL PLAIN; the user name and password are accepted whatever they are.
• The default virtual host / does not exist here; the virtual host must be a tenant identifier.
• No management HTTP API, no Streams, no plugins (MQTT, STOMP, shovel, federation).
• Public service only: publishing to a queue name through the default exchange, consuming from it, basic.get and a passive queue.declare create the queue if it does not exist, so they never fail with 404 NOT_FOUND. Queues and exchanges you declare count towards the tenant's 100 entities.

Messages from other interfaces

Messages sent through the other interfaces into the same queue arrive with their bodies as sent; a typed Service Bus body (an AMQP value such as a map) arrives as JSON text with the header x-pq-body-encoding = json. SQS message attributes, Pub/Sub attributes and Service Bus application properties arrive as headers, typed (strings, numbers, byte arrays). Headers whose names start x-pq- are the service's metadata. Your own decimal, table, array and void headers reach the other interfaces as text or JSON.

The full rules, with examples: Using one queue from several interfaces.

Errors

AMQP 0-9-1 frames are not counted against the per-address request budget; the connection and channel limits bound them instead.