amqp

Portable AMQP 0-9-1 (Advanced Message Queuing Protocol) client implementation. This library uses the sockets library and supports all backend Prolog systems supported by that library: ECLiPSe, GNU Prolog, SICStus Prolog, SWI-Prolog, Trealla Prolog, and XVM.

https://www.rabbitmq.com/amqp-0-9-1-reference.html

AMQP 0-9-1 is a binary wire-level protocol for message-oriented middleware. It is the de facto standard for message broker interoperability and is supported by RabbitMQ, Apache Qpid, Apache ActiveMQ, and other brokers.

API documentation

Open the ../../apis/library_index.html#amqp link in a web browser.

Loading

To load this library, load the loader.lgt file:

| ?- logtalk_load(amqp(loader)).

Testing

To test this library predicates, load the tester.lgt file:

| ?- logtalk_load(amqp(tester)).

Note: Integration tests require a running AMQP 0-9-1 server (e.g., RabbitMQ). RabbitMQ AMQP listens on port 5672 by default with guest/guest credentials.

Features

• Full AMQP 0-9-1 protocol support with binary frame encoding/decoding

• Connection management with heartbeat negotiation

• Automatic reconnection with configurable retry attempts and delays

• Connection pooling with automatic connection management

• Multiple concurrent channels over a single connection

• Exchange operations: declare, delete, bind, unbind

• Queue operations: declare, delete, bind, unbind, purge

• Basic messaging: publish, consume, get, ack, nack, reject

• Quality of service (QoS) with prefetch settings

• Transaction support: tx.select, tx.commit, tx.rollback

• Publisher confirms (RabbitMQ extension)

• SASL PLAIN authentication

• Custom message properties and headers

• Content properties: content-type, delivery-mode, priority, etc.

Usage

Connecting to an AMQP server

Basic connection to RabbitMQ with default settings:

?- amqp::connect(localhost, 5672, Connection, []).

Connection with custom credentials and virtual host:

?- amqp::connect(localhost, 5672, Connection, [
    username('myuser'),
    password('mypassword'),
    virtual_host('/myvhost'),
    heartbeat(30)
]).

Connection with automatic reconnection enabled:

?- amqp::connect(localhost, 5672, Connection, [
    reconnect(true),
    reconnect_attempts(5),
    reconnect_delay(2)
]).

This will attempt to connect up to 5 times with a 2 second delay between attempts. If all attempts fail, an amqp_error(reconnect_failed) error is thrown. The reconnection options are:

• reconnect(Boolean) - Enable automatic reconnection (default: false)

• reconnect_attempts(N) - Maximum number of connection attempts (default: 3)

• reconnect_delay(Seconds) - Delay between attempts in seconds (default: 1)

Opening a channel

AMQP operations require a channel. Open one on the connection:

?- amqp::channel_open(Connection, 1, Channel).

You can open multiple channels (with different numbers) on a single connection.

Declaring exchanges and queues

Declare a direct exchange:

?- amqp::exchange_declare(Channel, 'my.exchange', [
    type(direct),
    durable(true)
]).

Declare a queue:

?- amqp::queue_declare(Channel, 'my.queue', [
    durable(true),
    auto_delete(false)
]).

Declare a queue with server-generated name:

?- amqp::queue_declare(Channel, Queue, [exclusive(true)]).
% Queue will be unified with the generated name

Bind a queue to an exchange:

?- amqp::queue_bind(Channel, 'my.queue', 'my.exchange', [
    routing_key('my.routing.key')
]).

Publishing messages

Publish a simple message:

?- amqp::basic_publish(Channel, 'my.exchange', 'Hello, World!', [
    routing_key('my.routing.key')
]).

Publish with properties:

?- amqp::basic_publish(Channel, 'my.exchange', '{"data": "json"}', [
    routing_key('my.routing.key'),
    content_type('application/json'),
    delivery_mode(2),  % persistent
    correlation_id('abc123'),
    reply_to('reply.queue')
]).

Publish with custom headers:

?- amqp::basic_publish(Channel, 'my.exchange', 'Message', [
    routing_key('my.key'),
    headers([
        'x-custom-header'-longstr('value'),
        'x-priority'-int(5)
    ])
]).

Consuming messages

Start a consumer:

?- amqp::basic_consume(Channel, 'my.queue', [
    consumer_tag('my-consumer'),
    no_ack(false)
]).

Receive messages:

?- amqp::receive(Channel, Message, [timeout(5000)]).

Extract message data:

?- amqp::message_body(Message, Body),
   amqp::message_delivery_tag(Message, DeliveryTag),
   amqp::message_property(Message, content_type, ContentType).

Acknowledge a message:

?- amqp::basic_ack(Channel, DeliveryTag, []).

Reject a message:

?- amqp::basic_reject(Channel, DeliveryTag, [requeue(true)]).

Synchronous get

Get a single message synchronously:

?- amqp::basic_get(Channel, 'my.queue', [no_ack(false)]).

Quality of Service

Set prefetch count to limit unacknowledged messages:

?- amqp::basic_qos(Channel, [prefetch_count(10)]).

Transactions

Enable transactions on a channel:

?- amqp::tx_select(Channel).

Publish messages within a transaction:

?- amqp::basic_publish(Channel, 'exchange', 'Msg1', [routing_key('key')]),
   amqp::basic_publish(Channel, 'exchange', 'Msg2', [routing_key('key')]),
   amqp::tx_commit(Channel).

Rollback a transaction:

?- amqp::tx_rollback(Channel).

Publisher Confirms (RabbitMQ extension)

Enable publisher confirms:

?- amqp::confirm_select(Channel).

Closing connections

Close a channel:

?- amqp::channel_close(Channel).

Close the connection:

?- amqp::close(Connection).

Connection Pooling

The library provides connection pooling through the amqp_pool category. To create a connection pool, define an object that imports this category:

Defining a pool

:- object(my_pool,
    imports(amqp_pool)).
:- end_object.

Initializing the pool

Initialize the pool with configuration options:

?- my_pool::initialize([
    host(localhost),
    port(5672),
    min_size(2),
    max_size(10),
    connection_options([
        username('guest'),
        password('guest'),
        virtual_host('/')
    ])
]).

Pool configuration options:

• host(Host) - AMQP server hostname (default: localhost)

• port(Port) - AMQP server port (default: 5672)

• min_size(N) - Minimum connections to maintain (default: 1)

• max_size(N) - Maximum connections allowed (default: 10)

• connection_options(Options) - Options passed to amqp::connect/4 (default: [])

Acquiring and releasing connections

Manually acquire and release connections:

?- my_pool::acquire(Connection),
   amqp::channel_open(Connection, 1, Channel),
   % ... use the channel ...
   amqp::channel_close(Channel),
   my_pool::release(Connection).

Using with_connection/1

Use with_connection/1 for automatic connection management:

?- my_pool::with_connection(do_work).

do_work(Connection) :-
    amqp::channel_open(Connection, 1, Channel),
    amqp::basic_publish(Channel, '', 'Hello!', [routing_key('my.queue')]),
    amqp::channel_close(Channel).

The connection is automatically released even if the goal fails or throws an exception.

Pool statistics

Get pool statistics:

?- my_pool::stats(stats(Available, InUse, Total, MinSize, MaxSize)).

Resizing the pool

Resize the pool at runtime:

?- my_pool::resize(5, 20).

Destroying the pool

Close all connections and clear pool state:

?- my_pool::destroy.

Creating pools dynamically

Pools can also be created at runtime using create_object/4:

?- create_object(dynamic_pool, [imports(amqp_pool)], [], []),
   dynamic_pool::initialize([host(localhost), port(5672)]).

Binary Frame Encoding

AMQP 0-9-1 is a binary protocol. The library provides low-level encoding and decoding predicates for working with raw frames:

Frame structure

An AMQP frame consists of:

• Type (1 byte): 1=method, 2=header, 3=body, 8=heartbeat

• Channel (2 bytes): Channel number (0 for connection frames)

• Size (4 bytes): Payload size

• Payload (Size bytes): Frame-specific data

• Frame end (1 byte): 0xCE marker

Encoding/decoding frames

Encode a frame to bytes:

?- amqp::encode_frame(Frame, Bytes).

Decode bytes to a frame:

?- amqp::decode_frame(Bytes, Frame).

Data Types

The library handles AMQP data types automatically:

• octet: 8-bit unsigned integer

• short: 16-bit unsigned integer (big-endian)

• long: 32-bit unsigned integer (big-endian)

• longlong: 64-bit unsigned integer (big-endian)

• shortstr: Short string (length <= 255)

• longstr: Long string

• table: Field table (key-value pairs)

• array: Field array

Field values in tables use type tags:

• bool(true/false) - Boolean

• byte(V) - Signed 8-bit

• short(V) - Signed 16-bit

• int(V) - Signed 32-bit

• long(V) - Signed 64-bit

• float(V) - 32-bit float

• double(V) - 64-bit float

• longstr(V) - Long string

• table(Pairs) - Nested table

• array(Values) - Array

• timestamp(V) - Timestamp

• void - No value

Float/Double Encoding/Decoding

The library implements proper IEEE 754 encoding and decoding for floating-point values:

Float (32-bit IEEE 754 single precision):

• Format: 1 sign bit + 8 exponent bits (bias 127) + 23 mantissa bits

• Encoded as 4 bytes in big-endian order

• Used for float(Value) field values

Double (64-bit IEEE 754 double precision):

• Format: 1 sign bit + 11 exponent bits (bias 1023) + 52 mantissa bits

• Encoded as 8 bytes in big-endian order

• Used for double(Value) field values

Special values representation:

The library represents IEEE 754 special values using Prolog compound terms:

• @infinity - Positive infinity

• @negative_infinity - Negative infinity

• @not_a_number - NaN (Not a Number)

These special values are automatically encoded to and decoded from their standard IEEE 754 binary representations.

Known limitation:

Most Prolog backends cannot distinguish between -0.0 and 0.0 when comparing floating-point values. While the library correctly encodes and decodes both positive and negative zero according to IEEE 754, these values will typically compare as equal in Prolog arithmetic operations. This limitation is unlikely to affect typical AMQP messaging applications.

Error Handling

The library throws structured errors:

• amqp_error(connection_failed) - TCP connection failed

• amqp_error(auth_failed) - Authentication failed

• amqp_error(protocol_error(Msg)) - Protocol violation

• amqp_error(channel_error(Msg)) - Channel-level error

• amqp_error(exchange_error(Msg)) - Exchange operation failed

• amqp_error(queue_error(Msg)) - Queue operation failed

• amqp_error(basic_error(Msg)) - Basic operation failed

• amqp_error(tx_error(Msg)) - Transaction error

Comparison with STOMP

AMQP 0-9-1 and STOMP are both messaging protocols, but differ significantly:

Feature	AMQP 0-9-1	STOMP
Protocol type	Binary	Text
Complexity	High	Low
Exchange types	Multiple	Broker-dependent
Routing	Flexible	Simple
Transactions	Native	Native
QoS	Native prefetch	Limited
Performance	Higher	Lower

Use AMQP when you need:

• Fine-grained routing control

• High performance

• Advanced message patterns

• Exchange-based routing

Use STOMP when you need:

• Simple text-based protocol

• Easy debugging

• Protocol simplicity

Future Work

• Logtalk protocols for messaging patterns (request/reply, pub/sub, etc.)

• Categories for common message transformations

• Async message handlers

Known Limitations and Issues

• SSL/TLS connections not yet supported (use stunnel or similar)

• Heartbeat sending must be done manually via send_heartbeat/1

• Most Prolog backends cannot distinguish between -0.0 and 0.0 (see the “Float/Double Encoding/Decoding” section above for details)

• GNU Prolog support is partial as it doesn’t currently support null characters in atoms

AMQP 1.0 vs AMQP 0-9-1

This library implements AMQP 0-9-1 only. Despite the similar name, AMQP 1.0 is a fundamentally different protocol and is not supported by this library. The two versions are not wire-compatible and have different conceptual models:

Aspect	AMQP 0-9-1	AMQP 1.0
Design	Broker-centric	Peer-to-peer
Routing	Exchanges + queues	Link-based addressing
Frame structure	Custom binary framing	Layered performatives
Data encoding	Custom type system	CBOR-like encoding

Key differences:

• AMQP 0-9-1 uses exchanges, bindings, and queues as core protocol concepts

• AMQP 1.0 abstracts these as broker-specific “nodes” and uses links

• Frame encoding, handshake, and message format are completely different

• Minimal code could be shared between the two implementations

AMQP 1.0 support would require a separate library with its own distinct API reflecting the link-based model, rather than the exchange/queue model of AMQP 0-9-1.

Practical note: RabbitMQ and most message brokers continue to primarily use AMQP 0-9-1, making this library suitable for the vast majority of use cases.