pma/amqp: an Elixir client for RabbitMQ with a Langohr shaped API

A wrapper, and the two projects its API comes from

pma/amqp is described in the README as a simple Elixir wrapper for the Erlang RabbitMQ 3/4 client, with the protocol version given as AMQP 0.9.1. So the wire protocol and the broker compatibility come from a library that already exists, and what this project contributes is the Elixir surface.

The second influence is named too. The API is based on Langohr, a Clojure client for RabbitMQ. That is a design decision worth pausing on, because it tells you where the ergonomics come from. Langohr is a client whose shape is well suited to a functional language, with modules for queues, exchanges, channels and basic operations, and the naming here follows it: AMQP.Queue, AMQP.Exchange, AMQP.Basic, AMQP.Channel, AMQP.Connection.

So there are two inherited influences. The Erlang client provides the transport and the protocol. Langohr provides the shape of the API. An Elixir developer who has used Langohr will find the module layout familiar, and an Elixir developer coming from the raw Erlang client will find it a lot of names to learn.

The MIT licence covers the wrapper. The release history is steady rather than fast, with v4.3.0 on 2026-10-06, v4.2.1 on 2026-08-30 and v4.2.0 the same day, and the last push was on 2026-10-06. The primary language is Elixir, the default branch is main.

One more piece of the history is documented, and it is the kind of detail that only shows up in a project that has broken its API before: the README links a set of upgrade guides, for 0.x to 1.x, 1.x to 2.x, 2.x to 3.x and 3.x to 4.x. Four major version bumps, each with a breaking change guide, is a library whose API has moved more than once.

Opening a connection and a channel in two lines

The README's usage example is an interactive session, and it is the clearest way to see the shape of the API.

The first step is a connection, opened with AMQP.Connection.open, which returns an ok tuple wrapping a struct with the process id of the connection.

iex> {:ok, conn} = AMQP.Connection.open()
# {:ok, %AMQP.Connection{pid: #PID<0.165.0>}}

The second is a channel on that connection.

iex> {:ok, chan} = AMQP.Channel.open(conn)
# {:ok, %AMQP.Channel{conn: %AMQP.Connection{pid: #PID<0.165.0>}, pid: #PID<0.177.0>}

The struct fields make the process model visible. A connection is a process, and a channel is another process that holds a reference to its connection, so the whole thing is supervised by the Erlang process model rather than by a handle you have to close by hand.

From there the operations read like the AMQP specification. A queue is declared with AMQP.Queue.declare, an exchange with AMQP.Exchange.declare, and the two are connected with AMQP.Queue.bind. Publishing is AMQP.Basic.publish, and pulling a single message is AMQP.Basic.get, which returns a payload and metadata tuple.

iex> AMQP.Queue.declare(chan, "test_queue")
# {:ok, %{consumer_count: 0, message_count: 0, queue: "test_queue"}}

Consuming is a callback passed to AMQP.Queue.subscribe, and the return value carries the consumer tag the broker assigned.

iex> AMQP.Queue.subscribe(chan, "test_queue", fn payload, _meta -> IO.puts("Received: #{payload}") end)
# {:ok, "amq.ctag-5L8U-n0HU5doEsNTQpaXWg"}

A GenServer consumer, and the three broker callbacks it must handle

The second half of the README example is a full consumer built as a GenServer, and it is the most instructive part of the documentation because it shows what the library expects from your code.

The setup function opens a connection with a URL, opens a channel, sets up the queue, applies a prefetch count with Basic.qos, and registers the process as a consumer with Basic.consume.

    # Limit unacknowledged messages to 10.
    :ok = Basic.qos(chan, prefetch_count: 10)
    # Register the GenServer process as a consumer.
    {:ok, _consumer_tag} = Basic.consume(chan, @queue)

Then there are three handle_info clauses for the three messages the broker sends, and the reason for each is in the comments. A basic_consume_ok is the confirmation sent by the broker after registering the process as a consumer, and it is answered with a no-reply so the GenServer keeps running. A basic_cancel is sent when the consumer is unexpectedly cancelled, such as after a queue deletion, and the clause stops the process with a normal reason. A basic_cancel_ok is the confirmation sent by the broker to the consumer process after a Basic.cancel.

The fourth clause handles the actual delivery, matching on the payload, the delivery tag and the redelivered flag.

  def handle_info({:basic_deliver, payload, %{delivery_tag: tag, redelivered: redelivered}}, chan) do
    # You might want to run payload consumption in separate Tasks in production
    consume(chan, tag, redelivered, payload)
    {:noreply, chan}
  end

The comment about running consumption in separate Tasks is a warning about doing slow work in the consumer process, and the reason the prefetch count is set to 10 is the same concern from the other side, bounding how many unacknowledged messages can be in flight.

The consume function shows the acknowledge, reject and dead letter cycle. A payload that converts to a number of 10 or less is acknowledged with Basic.ack, and a larger one is rejected with requeue set to false. The rescue clause rejects with requeue set to the negation of redelivered, so a message is retried once and then given up on, and the comment explains that you might also want to catch exit signals in production code. The last comment is the one to remember: make sure you call ack, nack, or reject, otherwise the consumer will stop receiving messages.

Config driven connections that open, and reconnect, on their own

The Configuration section describes an option that removes a lot of the boilerplate the example code shows.

You can define a connection and channel in your config, and AMQP will automatically open the connection and channel when the application starts, and automatically try to reconnect if they are disconnected.

config :amqp,
  connections: [
    myconn: [url: "amqp://guest:guest@myhost:12345"],
  ],
  channels: [
    mychan: [connection: :myconn]
  ]

The named connection and channel are then reachable through AMQP.Application.

iex> {:ok, chan} = AMQP.Application.get_channel(:mychan)
iex> :ok = AMQP.Basic.publish(chan, "", "", "Hello")

That is a supervised connection and channel with the supervision tree and the reconnect logic in the library rather than in your supervision callback, which for a message queue client is usually where you want it.

There is one condition attached to that convenience, and the README states it plainly. When a channel goes down and reconnects, you have to ensure your consumer subscribes to the channel again. The automatic reconnect restores the connection and the channel, but it cannot restore your subscription, because the subscription lives in your code. So a supervised channel removes the connection boilerplate and leaves the subscription problem in your hands, which is a fair trade but one to know about in advance.

The rest of the configuration is broker side. The repository ships compose files for a local broker, and the plain docker-compose.yml brings up a rabbitmq:alpine service with the AMQP port published as 5672.

What AMQP 0.9.1 means, and where the boundary is

The protocol version is the single most consequential fact about this client, and it is stated in the first line of the description: the Erlang RabbitMQ 3/4 client, AMQP 0.9.1.

That covers the two current RabbitMQ major versions, since the underlying client is described as the RabbitMQ 3/4 client and the compose files in the repository are separated into docker-compose.rabbitmq3.yml and docker-compose.rabbitmq4.yml. What it does not cover is AMQP 1.0, the other protocol RabbitMQ speaks, and there is nothing in the README suggesting a path to it. A project that needs 1.0 has to look elsewhere.

The second limit is the wrapper's own nature. This is a wrapper, not a reimplementation, so the surface area is bounded by what the Erlang client exposes and what Langohr's shape suggests. The README does not enumerate the full set of available operations, and it does not describe features like publisher confirms, transactions, or stream protocol support. Their absence from the documentation may mean they are unavailable, or it may mean they are simply undocumented, and the distinction matters to anyone who needs them, so the honest thing is to check the module documentation on Hex rather than assume.

A third limit is about the language, not the protocol. A library that is a wrapper around another library, whose API was copied from a Clojure library, and which has been through four major version upgrades, is a library where the version you pin is the version whose behaviour you know. The upgrade guides exist precisely because the API moved, and a team upgrading from 2.x to 4.x is crossing two breaking changes and needs to read both guides.

Against that, the direct Erlang client is the alternative, and the difference in approach is simple. The Erlang client is the protocol implementation, with a lower level surface, and using it directly from Elixir means handling the channel lifecycle, reconnect behaviour and subscription restoration yourself. This library provides the Elixir API, the supervision-friendly configuration and the Langohr-shaped module layout, and in exchange you accept the version constraints of the wrapper and the 0.9.1 protocol.

Editorial conclusion

pma/amqp fits an Elixir application that publishes to or consumes from RabbitMQ and wants the Erlang client's coverage without its low level surface, and it does not fit a project that needs AMQP 1.0, since the wrapper is for the 0.9.1 protocol. Before upgrading across a major version, read the matching upgrade guide in the wiki for 0.x to 1.x, 1.x to 2.x, 2.x to 3.x or 3.x to 4.x, and if you use the config driven connections, remember that a channel that reconnects requires your consumer to subscribe again.

Frequently asked questions

Official sources

1. Issues
2. License: MIT
3. pma/amqp on GitHub
4. README
5. Releases