amqp - AMQP 0-9-1 client (RabbitMQ)

Import with import "amqp.j" as amqp;. A client for RabbitMQ and compatible AMQP 0-9-1 brokers over net: connect, declare a queue, publish messages, and pull them back. The binary frame and method encoding is built by hand from bytes and the bitwise operators - the largest protocol module in the library. Needs the default jennifer binary. A protocol error or dropped connection throws Error{kind: "amqp"}.

import "amqp.j" as amqp;

def c as amqp.Conn init amqp.connect(amqp.options("localhost", "guest", "guest"));
amqp.declareQueue($c, "jobs", true);
amqp.publishText($c, "", "jobs", "hello");

def m as amqp.Message init amqp.get($c, "jobs", false);
if (not $m.empty) {
    amqp.ack($c, $m.deliveryTag);
}
amqp.close($c);

Runnable: examples/modules/amqp_demo.j.

Connecting

connect runs the full handshake - protocol header, Connection.Start / Start-Ok (SASL PLAIN auth), Tune / Tune-Ok (heartbeats disabled), Open / Open-Ok, then Channel.Open - and returns a Conn on a single channel.

def struct amqp.Options { host as string, port as int, user as string, password as string, vhost as string };

Queues and publishing

declareQueue returns QueueInfo{name, messageCount, consumerCount}. publish sends the method frame, a content-header frame (body size), and a body frame. Use exchange "" (the default exchange) to route straight to a queue by name via routingKey.

Consuming (pull)

amqp.get(c, queue, autoAck) pulls the next message with Basic.Get - a synchronous pull, not an async delivery loop. Call it in a loop until Message.empty is true; ack each message (unless autoAck).

def struct amqp.Message {
    empty as bool,        # true when the queue was empty (other fields zero)
    deliveryTag as int,   # pass to ack
    exchange as string,
    routingKey as string,
    body as bytes
};

def more as bool init true;
repeat {
    def m as amqp.Message init amqp.get($c, "jobs", false);
    if ($m.empty) {
        $more = false;
    } else {
        # handle $m.body
        amqp.ack($c, $m.deliveryTag);
    }
} until (not $more);

Scope

• Pull, not push. Receiving is Basic.Get (one message per call); streaming Basic.Consume with server-pushed Basic.Deliver (an async loop) is a follow-on.
• One channel, no publisher confirms / transactions. A single channel (1) is opened; publish is fire-and-forget (no Confirm.Select).
• No message properties. Publishes carry an empty property set (no content-type, headers, or persistence flag on the message itself - queue durability is set at declareQueue).
• SASL PLAIN only, no TLS (amqps) in this version - use a trusted network or a local broker.
• The largest protocol module. If the tree-walker ever becomes the bottleneck for high-throughput messaging, this is a candidate to reimplement as a Go library.

See also

• net.md - the TCP transport this is built on.
• mqtt.md / redis.md - the other binary-protocol clients over net.
• modules/index.md - the module catalog and import rules.