Elixir CICodacy Badge

🐇 Elixir RabbitMQ Client

rabbit_mq is an opinionated RabbitMQ client to help you build balanced and consistent Consumers and Producers.

Table of contents

Installation and Usage

Add :rabbit_mq as a dependency to your project's mix.exs:

defp deps do
  [
    {:rabbit_mq, "~> 0.0.0-alpha-8"}
  ]
end

Documentation

The full documentation is published on hex.

The following modules are provided;

Sample usage

⚠️ The following examples assume you've already set up your (RabbitMQ) routing topology as shown below.

ℹ️ Consult the RabbitMQ.Topology module to learn how to quickly establish desired routing topology.

source_name source_kind destination_name destination_kind routing_key arguments
customer exchange customer/customer.created queue customer.created []
customer exchange customer/customer.updated queue customer.updated []

As seen in the RabbitMQ Management dashboard:

RabbitMQ Topology

First, ensure you point to a valid amqp_url by configuring :rabbit_mq in your config.exs.

config :rabbit_mq, :amqp_url, "amqp://guest:guest@localhost:5672"

ℹ️ For advanced configuration options, consult the Configuration section.

Producers

Let's define our CustomerProducer first. We will use this module to publish messages onto the "customer" exchange.

defmodule RabbitSample.CustomerProducer do
  @moduledoc """
  Publishes pre-configured events onto the "customer" exchange.
  """

  use RabbitMQ.Producer, exchange: "customer", worker_count: 3

  @doc """
  Publishes an event routed via "customer.created".
  """
  def customer_created(customer_id) when is_binary(customer_id) do
    opts = [
      content_type: "application/json",
      correlation_id: UUID.uuid4(),
      mandatory: true
    ]

    payload = Jason.encode!(%{v: "1.0.0", customer_id: customer_id})

    publish(payload, "customer.created", opts)
  end

  @doc """
  Publishes an event routed via "customer.updated".
  """
  def customer_updated(updated_customer) when is_map(updated_customer) do
    opts = [
      content_type: "application/json",
      correlation_id: UUID.uuid4(),
      mandatory: true
    ]

    payload = Jason.encode!(%{v: "1.0.0", customer_data: updated_customer})

    publish(payload, "customer.updated", opts)
  end
end

⚠️ Please note that all Producer workers implement "reliable publishing". Each Producer worker handles its publisher confirms asynchronously, striking a delicate balance between performance and reliability.

To understand why this is important, please refer to the reliable publishing implementation guide.

ℹ️ In the unlikely event of an unexpected Publisher nack, your server will be notified via the on_unexpected_nack/2 callback, letting you handle such exceptions in any way you see fit.

Consumers

To consume messages off the respective queues, we will define 2 separate consumers.

⚠️ Please note that automatic message acknowledgement is disabled in rabbit_mq, therefore it's your responsibility to ensure messages are ack'd or nack'd.

ℹ️ Please consult the Consumer Acknowledgement Modes and Data Safety Considerations for more details.

defmodule RabbitSample.CustomerCreatedConsumer do
  use RabbitMQ.Consumer, queue: "customer/customer.created", worker_count: 2, prefetch_count: 3

  require Logger

  def consume(payload, meta, channel) do
    Logger.info("Customer #{payload} created.")
    ack(channel, meta.delivery_tag)
  end
end
defmodule RabbitSample.CustomerUpdatedConsumer do
  use RabbitMQ.Consumer, queue: "customer/customer.updated", worker_count: 2, prefetch_count: 6

  require Logger

  def consume(payload, meta, channel) do
    Logger.info("Customer updated. Data: #{payload}.")
    ack(channel, meta.delivery_tag)
  end
end

Use under supervision tree

And finally, we will start our application.

ℹ️ To run RabbitMQ locally, see our docker-compose.yaml for a sample Docker Compose set up.

defmodule RabbitSample.Application do
  use Application

  def start(_type, _args) do
    children = [
      RabbitSample.CustomerProducer,
      RabbitSample.CustomerCreatedConsumer,
      RabbitSample.CustomerUpdatedConsumer
    ]

    opts = [strategy: :one_for_one, name: RabbitSample.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

Using iex;

iex -S mix

The resulting application topology should look like this:

Application Topology

Upon closer inspection using the RabbitMQ Management dashboard, we see that:

Connections

ℹ️ Detailed view of how individual workers have set up their channels. Note that the different prefetch counts correspond to the different configuration we provided in our Consumers, and that the Producer's 3 worker channels operate in Confirm mode.

Channels

Configuration

The following options can be configured.

config :rabbit_mq,
  amqp_url: "amqp://guest:guest@localhost:5672",
  heartbeat_interval_sec: 60,
  reconnect_interval_ms: 2500,
  max_channels_per_connection: 16

⚠️ Please consult the Channels Resource Usage guide to understand how to best configure :max_channels_per_connection.

⚠️ Please consult the Detecting Dead TCP Connections with Heartbeats and TCP Keepalives guide to understand how to best configure :heartbeat_interval_sec.

Balanced performance and reliability

The RabbitMQ modules are pre-configured with sensible defaults and follow design principles that improve and delicately balance both performance and reliability.

This has been possible through

⚠️ While most of the heavy-lifting is provided by the library itself, reading through the documents below before running any application in production is thoroughly recommended.

TODO

A quick and dirty tech-debt tracker, used in conjunction with Issues.