Lepus.Consumer behaviour (Lepus v0.1.6)

Consumer receives messages from a RabbitMQ queue. It uses Broadway and BroadwayRabbitMQ.Producer under the hood.

Features

Optionally retries failed messages with exponential backoff
Optionally publishes failed messages to separate queue
Supports RPC

Topology

┌►   [queue]
│       │
│       ▼
│ (Lepus.Consumer) ───► {failed exchange} ───► [failed queue]
│       │
│       ▼
│ {delay exchange}
│       │
│       ▼
│ [retry queue]
│       │
│       ▼
└ {retry exchange}

Example

Define a consumer:

defmodule MyApp.MyConsumer do
  @behaviour Lepus.Consumer

  @impl Lepus.Consumer
  def options do
    [
      connection: [
        host: "localhost",
        port: 5672,
        virtual_host: "/",
        username: "guest",
        password: "guest"
      ],
      exchange: "my_exchange",
      routing_key: "my_routing_key",
      queue: "my_queue",
      store_failed: true,
      max_retry_count: 5
    ]
  end

  @impl Lepus.Consumer
  def handle_message(data, metadata) do
    Do.something_with(data, metadata)
  end
end

Then add it to your supervision tree (usually in lib/my_app/application.ex):

children = [
  {Lepus.Consumer, consumers: [MyApp.MyConsumer]}
]

children |> Supervisor.start_link(strategy: :one_for_one)

You can also define some global options for all the consumers in the supervision tree:

children = [
  {Lepus.Consumer,
    consumers: [MyApp.MyConsumer, MyApp.OtherConsumer],
    options: [connection: rabbit_connection, exchange: "my_exchange"]}
]

children |> Supervisor.start_link(strategy: :one_for_one)

Summary

Types

metadata()

payload()

Callbacks

handle_message(payload, metadata)

Receives message payload and metadata.

options()

Should return keyword list of options (if defined). You can also define options in the supervision tree as described above.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

start_link(opts)

Types

metadata()

@type metadata() :: %{
  rpc: boolean(),
  retry_count: non_neg_integer(),
  retriable: boolean(),
  client: String.t() | nil,
  rabbit_mq_metadata: map()
}

payload()

@type payload() :: any()

Callbacks

handle_message(payload, metadata)

@callback handle_message(payload(), metadata()) ::
  :ok
  | {:ok, payload()}
  | :retry
  | {:retry, payload()}
  | :error
  | {:error, payload()}

Receives message payload and metadata.

`payload`

Usually it's a binary RabbitMQ message payload. But it could be parsed JSON (map, list, etc.) if content type of the message is "application/json" or it was sent via Lepus.publish_json/4

`metadata`

It's a map with the keys:

rpc - Defines if the message was sent using RPC pattern (reply_to and correlation_id is not empty). The function should return {:ok, result} if rpc is true.
retry_count - Count of retries if the message was retried. Otherwise 0.
retriable – Defines if the message will be retried in case of error.
client – "lepus" if the message was sent via Lepus.publish/4 or Lepus.publish_json/4.
rabbit_mq_metadata - metadata field from BroadwayRabbitMQ.Producer.

Should return in case of using RPC:

{:ok, response} – so Lepus.publish/4 or Lepus.publish_json/4 returns {:ok, response}
{:error, response} – so Lepus.publish/4 or Lepus.publish_json/4 returns {:error, response}

Should return in case of not using RPC:

:ok – exists successfully.
:error – exits with error. The message could be retried if max_retry_count > 0.
{:error, error_message} – The same as :error but adds the error_message to RabbitMQ headers.
:retry – always retries message (even if max_retry_count == 0).
{:retry, reason} – The same as :retry but adds the reason to RabbitMQ headers.

options()

(optional)

@callback options() :: Keyword.t()

Should return keyword list of options (if defined). You can also define options in the supervision tree as described above.

Options

:connection - Required. Defines an AMQP URI or a set of options used by the RabbitMQ client to open the connection with the RabbitMQ broker. See AMQP.Connection.open/1 for the full list of options.
:exchange (String.t/0) - Required. The name of the RabbitMQ exchange.
:routing_key (String.t/0) - Required. The name of the RabbitMQ routing_key.
:delay_exchange (String.t/0) - Redefines delay exchange name. The default value is "#{exchange}.delay"
:retry_exchange (String.t/0) - Redefines retry exchange name. The default value is "#{exchange}.retry"
:failed_exchange (String.t/0) - Redefines error exchange name. The default value is "#{exchange}.failed"
:queue (String.t/0) - Redefines queue name. The default value is "#{exchange}.#{routing_key}"
:retry_queue (String.t/0) - Redefines retry queue name. The default value is "#{queue}.retry"
:failed_queue (String.t/0) - Redefines error queue name. The default value is "#{queue}.failed"
:processor (keyword/0) - Broadway processor options. Only one processor is used. See Broadway.start_link/2 for the full list of options. The default value is [].
:max_retry_count (non_neg_integer/0) - The maximum count of message retries. The default value is 0.
:store_failed (boolean/0) - Defines if the failed message should be published to failed_queue. The default value is false.
:queues_type - Defines queues type (one of ["classic", "quorum"]). Used during queue declaration. The default value is "classic".

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

start_link(opts)

@spec start_link(keyword()) :: Supervisor.on_start()