gen_rmq v0.1.7 GenRMQ.Consumer behaviour

A behaviour module for implementing the RabbitMQ consumer.

It will:

setup RabbitMQ connection / channel and keep them in a state
create (if does not exist) a queue and bind it to an exchange
create deadletter queue and exchange
handle reconnections
call handle_message callback on every message delivery

Summary

Functions

ack(message)

Acknowledges given message

reject(message, requeue \\ false)

Requeues / rejects given message

start_link(module, options \\ [])

Starts GenRMQ.Consumer process with given callback module linked to the current process

stop(name, reason)

Synchronously stops the consumer with a given reason

Callbacks

consumer_tag()

Invoked to provide consumer tag

handle_message(message)

Invoked on message delivery

init()

Invoked to provide consumer configuration

Functions

ack(message)

Specs

ack(message :: GenRMQ.Message.t) :: :ok

Acknowledges given message

message - GenRMQ.Message struct

reject(message, requeue \\ false)

Specs

reject(message :: GenRMQ.Message.t, requeue :: Boolean.t) :: :ok

Requeues / rejects given message

message - GenRMQ.Message struct

requeue - indicates if message should be requeued

start_link(module, options \\ [])

Specs

start_link(module :: Module.t, options :: Keyword.t) ::
  {:ok, Pid.t} |
  {:error, Any.t}

Starts GenRMQ.Consumer process with given callback module linked to the current process

module - callback module implementing GenRMQ.Consumer behaviour

Options

:name - used for name registration

Return values

If the consumer is successfully created and initialized, this function returns {:ok, pid}, where pid is the PID of the consumer. If a process with the specified consumer name already exists, this function returns {:error, {:already_started, pid}} with the PID of that process.

Examples:

GenRMQ.Consumer.start_link(Consumer, name: :consumer)

stop(name, reason)

Specs

stop(name :: Atom.t | Pit.t, reason :: Any.t) :: :ok

Synchronously stops the consumer with a given reason

name - pid or name of the consumer to stop reason - reason of the termination

Examples:

GenRMQ.Consumer.stop(:consumer, :normal)

Callbacks

consumer_tag()

Specs

consumer_tag :: String.t

Invoked to provide consumer tag

Examples:

def consumer_tag() do
  "hostname-app-version-consumer"
end

handle_message(message)

Specs

handle_message(message :: GenRMQ.Message.t) :: :ok

Invoked on message delivery

message - GenRMQ.Message struct

Examples:

def handle_message(message) do
  # Do something with message and acknowledge it
  GenRMQ.Consumer.ack(message)
end

init()

Specs

init :: [queue: String.t, exchange: String.t, routing_key: String.t, prefetch_count: String.t, uri: String.t, concurrency: Boolean.t, queue_ttl: Integer.t, retry_delay_function: Function.t]

Invoked to provide consumer configuration

Return values

Mandatory:

uri - RabbitMQ uri

queue - the name of the queue to consume. If does not exist it will be created

exchange - the name of the exchange to which queue should be bind. If does not exist it will be created

routing_key - queue binding key

prefetch_count - limit the number of unacknowledged messages

Optional:

queue_ttl - controls for how long a queue can be unused before it is automatically deleted. Unused means the queue has no consumers, the queue has not been redeclared, and basic.get has not been invoked for a duration of at least the expiration period

concurrency - defines if handle_message callback is called in seperate process using spawn function. By default concurrency is enabled. To disable, set it to false

retry_delay_function - custom retry delay function. Called when the connection to the broker cannot be established. Receives the connection attempt as an argument (>= 1) and is expected to wait for some time. With this callback you can for example do exponential backoff. The default implementation is a linear delay starting with 1 second step.

Examples:

def init() do
  [
    queue: "gen_rmq_in_queue",
    exchange: "gen_rmq_exchange",
    routing_key: "#",
    prefetch_count: "10",
    uri: "amqp://guest:guest@localhost:5672",
    concurrency: true,
    queue_ttl: 5000,
    retry_delay_function: fn attempt -> :timer.sleep(1000 * attempt) end
  ]
end