AMQP Helpers

Non opinionated AMQP helpers for common scenarios.

Installation

To use AMQPHelpers you need AMQP. You can install both by adding amqp and amqp_helpers to your list of dependencies in mix.exs:

def deps do
  [
    {:amqp, "~> 4.0"},
    {:amqp_helpers, "~> 1.1"}
  ]
end

Motivation

This library provides several wrappers and utilities around some common use cases of AMQP. It provides a simple interface, following common OTP patterns and Elixir library guidelines to avoid any opinionated interface. Other generalist abstractions can be built on top of AMQPHelpers to provide ergonomics or any other feature not strictly tied to AMQP.

Right now, the utilities built in this library are suited for uses that try to optimize AMQP for high throughput and for scenarios that require data safety at any cost.

Comparisons With Other Libraries

This is how this library relates to other available libraries in order to highlight the motivation behind this library:

AMQP: this library is not a replacement in any way to this library, but a companion.
AMQPx: in some way, similar to this library, but not tied to any particular use case and some of its functionality has been superseded by AMQP v2.x.
Broadway: this is an abstraction for message processing, which cannot be compared directly to this library. broadway_rabbitmq is a connector for AMQP and can be adapted to use this library if needed. Nevertheless, unless you want to support multiple transports, most of the features provided by Broadway can be implemented directly by using some RabbitMQ/AMQP features (there are exceptions like graceful shutdown).
GenAmqp: provides some utilities in the same way that this library, but not tied to any specific use case. Also, some of its functionality has been superseded by AMQP v2.x.
PlugAmqp: It will use this library, but have different purposes (it implements RPC pattern over AMQP).
Rambla: similar to Broadway, but from the publishing point of view.

In summary, this library provides helpers tied to specific use cases of AMQP, without any kind of abstraction over it. If you are looking to support different kinds of transports in your library, check out libraries like Broadway or Rambla.

AMQP Good Practices

This library enforces some good practices that have no downside for any application using AMQP and ease the development of AMQP related features.

The first one is the use of an AMQP implementation behind a behaviour, called AMQPHelpers.Adapter. This allow us to provide stub, mocks or even different transport layers that mimic the AMQP interface.

The second one is the use of application connection/channels. This an AMQP v2.x feature which replaces (or aid) previous connections supervisors or channel pools. As general thumb rule, your application should have at most two connection (in/out) and one channel per multiplexing process.

User Case Scenarios

These are the AMQP use cases are covered in AMQP Helpers right now:

A performance-intensive scenario in which a high throughput message delivery rate is desired. Trade-offs in reliability are acceptable.
A reliable scenario in which data safety is a must, even if performance is compromised.
Remote procedure calls using Direct Reply-to.

There are some other features, like High Availability, Observability, Exclusivity, etc. that can be achieved in both scenarios but are not explicitly covered here.

High Throughput

To achieve the best performance in terms of message delivery some trade-off must be done, which usually impacts the reliability and/or coherence of the system.

Durability should be disabled. In other words, messages will not be persisted, so messages could be lost in a broker outage scenario. This can be configured by declaring queues as non-durable and publishing messages with persistent set to false.

Acknowledges, from any communication direction, should be disabled. This means that the publisher should not confirm deliveries, and consumers should not acknowledge messages. Messages could be lost on the flight because of network or edge issues. Publisher confirms are not enabled by default, so nothing has to be done in terms of publishing. Consuming requires disabling acknowledging, which can be done by setting the no_ack flag on.

The AMQPHelpers.HighThroughput module provides functions that enforce these requirements for publishing and consuming. These are simple wrappers around AMQP library functions. They add very little aside from being explicit about a feature of some scenario (performance intensive).

Some other considerations should have taken into account when declaring queues for this purpose, which are:

x-max-length or x-max-length-bytes with x-expires and x-messages-ttl should be used to avoid large queues. Large queues slowdown message delivery. This option limits the size of the queue to a known threshold.
x-queue-mode should not be "lazy" to avoid moving messages to disk.
Queues should not be replicated, nor enabling high availability nor using quorum queues. Both method have implications in the performance of the queue.
durable should be set to false.

Reliability

For reliability, Acknowledgements and Confirms are used, which have an impact in performance. Reliability is not only achieved at client level, broker configuration and monitoring are also important topics here. You can read RabbitMQ’s Reliability Guide for more information.

The AMQPHelpers.Reliability.Consumer and AMQPHelpers.Reliability.Producer provide processes to consume and publish messages in a reliable fashion. The Consumer uses standard AMQP acknowledges mechanism to notify that messages are consumed successfully. The Producer uses the Publisher Confirms extension to get notified about successfully message deliveries.

A system which uses the provided Reliability.Consumer and Reliability.Producer will guarantee that no message is lost, and at least one message is delivered. For high availability you can pair this processes with mirrored queues. For consistency, you can use deduplication plugin or quorum queues (Which also guarantees message order).

Testing

This library provides an AMQP interface at AMQPHelpers.Adapter which can be used with libraries like Mox to mock any part of the AMQP interface. Check out the tests of this library to see some examples.

Two adapter implementations are provided with this library:

AMQPHelpers.Adapters.Stub - A stub implementation that only logs calls.
AMQPHelpers.Adapters.AMQP - An implementation that uses AMQP library.

All the functionally exposed by the library support in one or another way a configurable adapter (defaults to AMQP implementation).