`amqp` API reference

An AMQP 0-9-1 client over net for RabbitMQ and compatible brokers. connect runs the connection + channel handshake (protocol header, Connection.Start / Start-Ok with SASL PLAIN auth, Tune / Tune-Ok, Open / Open-Ok, Channel.Open); declareQueue declares a classic queue and declareQuorumQueue a replicated quorum queue; publish sends a message (method + content-header + body frames); get pulls the next message with Basic.Get (a synchronous pull, no async delivery loop); ack acknowledges it; close shuts the connection down cleanly.

The binary frame and method encoding is built by hand from bytes and the bitwise operators - the largest protocol module here. Needs the default jennifer binary (net); a protocol error or dropped connection throws Error{kind: "amqp"}. Uses one channel (1); heartbeats are disabled.

Import with import "amqp.j" as amqp;. See the amqp guide for prose and examples.

Functions

`amqp.ack(c as Conn, deliveryTag as int)`

Acknowledge a delivered message by its tag.

Parameters

c {Conn} - the connection
deliveryTag {int} - the delivery tag from a got Message

`amqp.bindQueue(c as Conn, queue as string, exchange as string, routingKey as string)`

Bind a queue to an exchange with a routing key. For a "fanout" exchange the routing key is ignored; for "direct" it is matched exactly; for "topic" it is a pattern (* / #).

Parameters

c {Conn} - the connection
queue {string} - the queue name
exchange {string} - the exchange name
routingKey {string} - the binding routing key / pattern

Throws

{Error} - kind "amqp" on failure

`amqp.cancelConsume(c as Conn, consumerTag as string)`

Cancel a subscription started with consume (Basic.Cancel). After this the broker sends no further deliveries for that consumer tag.

Parameters

c {Conn} - the connection
consumerTag {string} - the consumer tag returned by consume

Throws

{Error} - kind "amqp" on failure

`amqp.close(c as Conn)`

Close the connection cleanly (Connection.Close) and shut the socket.

Parameters

c {Conn} - the connection

`amqp.confirmSelect(c as Conn)`

Put the channel into publisher-confirm mode (Confirm.Select). After this, each publish is confirmed by the broker; call waitConfirm once per publish to block for that confirmation. Idempotent.

Parameters

c {Conn} - the connection

Throws

{Error} - kind "amqp" on failure

`amqp.connect(opts as Options)`

Connect to a broker and open a channel.

Parameters

opts {Options} - the connection options

Returns {Conn} - the open connection

Throws

{Error} - kind "amqp" on a handshake failure

`amqp.consume(c as Conn, queue as string, autoAck as bool)`

Start a server-pushed subscription with Basic.Consume and return the consumer tag (the broker generates one). Follow with receiveDelivery in a loop - wrap it in a spawn so the pushed deliveries do not block the rest of the program. When autoAck is false, ack (or nack) each Delivery by its deliveryTag.

Parameters

c {Conn} - the connection
queue {string} - the queue name
autoAck {bool} - whether the broker should auto-acknowledge each delivery

Returns {string} - the consumer tag (pass to cancelConsume)

Throws

{Error} - kind "amqp" on failure

`amqp.declareExchange(c as Conn, name as string, exType as string)`

Declare an exchange (idempotent, durable). exType is "direct", "topic", or "fanout".

Parameters

c {Conn} - the connection
name {string} - the exchange name
exType {string} - the exchange type ("direct" / "topic" / "fanout")

Throws

{Error} - kind "amqp" on failure

`amqp.declareQueue(c as Conn, name as string, durable as bool)`

Declare a queue (idempotent). durable survives a broker restart.

Parameters

c {Conn} - the connection
name {string} - the queue name ("" for a server-generated name)
durable {bool} - whether the queue is durable

Returns {QueueInfo} - the queue name and message / consumer counts

Throws

{Error} - kind "amqp" on failure

`amqp.declareQuorumQueue(c as Conn, name as string)`

Declare a quorum queue: a Raft-replicated queue type for high availability and data safety. Sets the x-queue-type=quorum argument. Quorum queues are always durable (there is no durable flag) and cannot be server-named, so name must be non-empty. Re-declaring must use the same type, so do not also declare the same name as a classic queue.

Parameters

c {Conn} - the connection
name {string} - the queue name (non-empty)

Returns {QueueInfo} - the queue name and message / consumer counts

Throws

{Error} - kind "amqp" on an empty name or a declare failure

`amqp.get(c as Conn, queue as string, autoAck as bool)`

Pull the next message from a queue with Basic.Get. When autoAck is false, ack the returned message with ack.

Parameters

c {Conn} - the connection
queue {string} - the queue name
autoAck {bool} - whether the broker should auto-acknowledge

Returns {Message} - the message (its empty field is true when none was ready)

Throws

{Error} - kind "amqp" on failure

`amqp.nack(c as Conn, deliveryTag as int, requeue as bool)`

Negatively acknowledge a delivered message (Basic.Nack). With requeue true the broker re-queues the message for redelivery; with false it is dropped (or dead-lettered if configured). Nacks a single message (not "multiple").

Parameters

c {Conn} - the connection
deliveryTag {int} - the delivery tag from a Message / Delivery
requeue {bool} - whether the broker should re-queue the message

`amqp.options(host as string, user as string, password as string)`

Default options for a broker: port 5672, vhost "/".

Parameters

host {string} - the broker host
user {string} - the username
password {string} - the password

Returns {Options} - the options

`amqp.publish(c as Conn, exchange as string, routingKey as string, body as bytes)`

Publish a message body to an exchange with a routing key. Use exchange "" for the default exchange (routing key = queue name).

Parameters

c {Conn} - the connection
exchange {string} - the exchange name ("" for the default)
routingKey {string} - the routing key
body {bytes} - the message body

Throws

{Error} - kind "amqp" on failure

`amqp.publishText(c as Conn, exchange as string, routingKey as string, text as string)`

Publish a text message (UTF-8). Convenience over publish.

Parameters

c {Conn} - the connection
exchange {string} - the exchange name ("" for the default)
routingKey {string} - the routing key
text {string} - the message text

Throws

{Error} - kind "amqp" on failure

`amqp.publishWith(c as Conn, exchange as string, routingKey as string, body as bytes, props as Properties)`

Publish a message body carrying message properties (content-type, persistence, correlation-id, reply-to). Otherwise identical to publish. Build the property set with Properties{...}; an unset field ("" / false) is omitted from the wire.

Parameters

c {Conn} - the connection
exchange {string} - the exchange name ("" for the default)
routingKey {string} - the routing key
body {bytes} - the message body
props {Properties} - the message properties

Throws

{Error} - kind "amqp" on failure

`amqp.receiveDelivery(c as Conn)`

Block for the next server-pushed message (Basic.Deliver + content header + body frame(s)) and return it as a Delivery. Only valid after consume. This is the cooperative receive loop: there is no callback - the app calls receiveDelivery repeatedly, typically from inside its own spawn, and acts on each Delivery.

Parameters

c {Conn} - the connection

Returns {Delivery} - the next delivered message

Throws

{Error} - kind "amqp" on a non-delivery frame or a dropped connection

`amqp.waitConfirm(c as Conn)`

Block for the broker's confirmation of the next outstanding publish (only valid after confirmSelect). Returns true on Basic.Ack (the broker took responsibility for the message) and false on Basic.Nack (the broker could not).

Parameters

c {Conn} - the connection

Returns {bool} - true when confirmed, false when nacked

Throws

{Error} - kind "amqp" on an unexpected reply

`amqp.withPort(o as Options, port as int)`

Copy options with a different port.

Parameters

o {Options} - the options
port {int} - the port

Returns {Options} - a fresh options

`amqp.withVhost(o as Options, vhost as string)`

Copy options with a different virtual host.

Parameters

o {Options} - the options
vhost {string} - the virtual host

Returns {Options} - a fresh options

Structs

`amqp.Conn`

An open connection (single channel).

Field	Type	Description
`socket`	`net.Conn`	the underlying connection
`channel`	`int`	the channel number (always 1)
`frameMax`	`int`	the negotiated maximum frame size (bytes)

`amqp.Delivery`

A message pushed by the broker via Basic.Deliver (returned by receiveDelivery).

Field	Type	Description
`consumerTag`	`string`	the consumer tag the delivery is for
`deliveryTag`	`int`	the delivery tag (pass to ack / nack)
`redelivered`	`bool`	true if the broker has delivered this message before
`exchange`	`string`	the exchange the message came from
`routingKey`	`string`	the routing key
`body`	`bytes`	the message body

`amqp.Message`

A message pulled with get.

Field	Type	Description
`empty`	`bool`	true when the queue had no message (the other fields are zero)
`deliveryTag`	`int`	the delivery tag (pass to ack)
`exchange`	`string`	the exchange the message came from
`routingKey`	`string`	the routing key
`body`	`bytes`	the message body

`amqp.Options`

Connection options.

Field	Type	Description
`host`	`string`	the broker host
`port`	`int`	the broker port (5672 by default)
`user`	`string`	the username
`password`	`string`	the password
`vhost`	`string`	the virtual host ("/" by default)
`security`	`transport.Security`	`transport.Security.None` (plaintext AMQP, the default) or `.Tls` (AMQPS - TLS on connect, verifying the broker certificate); `.Starttls` is rejected (AMQP has no in-band upgrade)

`amqp.Properties`

Message properties carried on a publish (all optional; a "" string or a false flag omits that property from the wire, keeping the content header minimal).

Field	Type	Description
`contentType`	`string`	the MIME content-type (e.g. "application/json"); "" omits it
`persistent`	`bool`	true sets delivery-mode 2 (persistent); false omits it (transient)
`correlationId`	`string`	the correlation id (request / reply matching); "" omits it
`replyTo`	`string`	the reply-to queue / routing key; "" omits it

`amqp.QueueInfo`

Queue metadata returned by declareQueue.

Field	Type	Description
`name`	`string`	the queue name (the server's, for a server-named queue)
`messageCount`	`int`	the number of ready messages
`consumerCount`	`int`	the number of consumers

#amqp API reference

#Functions

#amqp.ack(c as Conn, deliveryTag as int)

#amqp.bindQueue(c as Conn, queue as string, exchange as string, routingKey as string)

#amqp.cancelConsume(c as Conn, consumerTag as string)

#amqp.close(c as Conn)

#amqp.confirmSelect(c as Conn)

#amqp.connect(opts as Options)

#amqp.consume(c as Conn, queue as string, autoAck as bool)

#amqp.declareExchange(c as Conn, name as string, exType as string)

#amqp.declareQueue(c as Conn, name as string, durable as bool)

#amqp.declareQuorumQueue(c as Conn, name as string)

#amqp.get(c as Conn, queue as string, autoAck as bool)

#amqp.nack(c as Conn, deliveryTag as int, requeue as bool)

#amqp.options(host as string, user as string, password as string)

#amqp.publish(c as Conn, exchange as string, routingKey as string, body as bytes)

#amqp.publishText(c as Conn, exchange as string, routingKey as string, text as string)

#amqp.publishWith(c as Conn, exchange as string, routingKey as string, body as bytes, props as Properties)

#amqp.receiveDelivery(c as Conn)

#amqp.waitConfirm(c as Conn)

#amqp.withPort(o as Options, port as int)

#amqp.withVhost(o as Options, vhost as string)

#Structs

#amqp.Conn

#amqp.Delivery

#amqp.Message

#amqp.Options

#amqp.Properties

#amqp.QueueInfo

`amqp` API reference

Functions

`amqp.ack(c as Conn, deliveryTag as int)`

`amqp.bindQueue(c as Conn, queue as string, exchange as string, routingKey as string)`

`amqp.cancelConsume(c as Conn, consumerTag as string)`

`amqp.close(c as Conn)`

`amqp.confirmSelect(c as Conn)`

`amqp.connect(opts as Options)`

`amqp.consume(c as Conn, queue as string, autoAck as bool)`

`amqp.declareExchange(c as Conn, name as string, exType as string)`

`amqp.declareQueue(c as Conn, name as string, durable as bool)`

`amqp.declareQuorumQueue(c as Conn, name as string)`

`amqp.get(c as Conn, queue as string, autoAck as bool)`

`amqp.nack(c as Conn, deliveryTag as int, requeue as bool)`

`amqp.options(host as string, user as string, password as string)`

`amqp.publish(c as Conn, exchange as string, routingKey as string, body as bytes)`

`amqp.publishText(c as Conn, exchange as string, routingKey as string, text as string)`

`amqp.publishWith(c as Conn, exchange as string, routingKey as string, body as bytes, props as Properties)`

`amqp.receiveDelivery(c as Conn)`

`amqp.waitConfirm(c as Conn)`

`amqp.withPort(o as Options, port as int)`

`amqp.withVhost(o as Options, vhost as string)`

Structs

`amqp.Conn`

`amqp.Delivery`

`amqp.Message`

`amqp.Options`

`amqp.Properties`

`amqp.QueueInfo`