Invoked to provide consumer tag

Examples:

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

Invoked when an error or timeout is encountered while executing handle_message callback

message - GenRMQ.Message struct reason - the information regarding the error

Examples:

To reject the message that caused the Task to fail you can do something like so:

def handle_error(message, reason) do
  # Do something with message and reject it
  Logger.warn("Failed to process message: #{inspect(message)}")

  GenRMQ.Consumer.reject(message)
end

The reason argument will either be the atom :killed if the Task timed out and needed to be stopped. Or it will be a 2 element tuple where the first element is the error stuct and the second element is the stacktrace:

{
  %RuntimeError{message: "Can't divide by zero!"},
  [
    {TestConsumer.ErrorWithoutConcurrency, :handle_message, 1, [file: 'test/support/test_consumers.ex', line: 98]},
    {GenRMQ.Consumer, :handle_message, 2, [file: 'lib/consumer.ex', line: 519]},
    {GenRMQ.Consumer, :handle_info, 2, [file: 'lib/consumer.ex', line: 424]},
    {:gen_server, :try_dispatch, 4, [file: 'gen_server.erl', line: 637]},
    {:gen_server, :handle_msg, 6, [file: 'gen_server.erl', line: 711]},
    {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 249]}
  ]
}

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

Invoked to provide consumer configuration

Return values

Mandatory:

connection - RabbitMQ connection options. Accepts same arguments as AMQP-library's Connection.open/2.

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

exchange - name or {type, name} of the exchange to which queue should be bound. If it does not exist, it will be created. For valid exchange types see GenRMQ.Binding.

routing_key - queue binding key, can also be a list.

prefetch_count - limit the number of unacknowledged messages.

Optional:

uri - RabbitMQ uri. Deprecated. Please use connection.

queue_options - queue options as declared in AMQP.Queue.declare/3.

If argument 'x-expires' is given to arguments, then it will be used instead of queue_ttl.

If argument 'x-max-priority' is given to arguments, then it will be used instead of queue_max_priority.

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

queue_max_priority - defines if a declared queue should be a priority queue. Should be set to a value from 1..255 range. If it is greater than 255, queue max priority will be set to 255. Values between 1 and 10 are recommended.

concurrency - defines if handle_message callback is called in separate process using supervised task. By default concurrency is enabled. To disable, set it to false

terminate_timeout - defines how long the consumer will wait for in-flight Tasks to complete before terminating the process. The value is in milliseconds and the default is 5_000 milliseconds.

handle_message_timeout - defines how long the handle_message callback will execute within a supervised task. The value is in milliseconds and the default is 5_000 milliseconds.

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.

reconnect - defines if consumer should reconnect on connection termination. By default reconnection is enabled.

deadletter - defines if consumer should setup deadletter exchange and queue. (Default: true).

deadletter_queue - defines name of the deadletter queue (Default: Same as queue name suffixed by _error).

deadletter_queue_options - queue options for the deadletter queue as declared in AMQP.Queue.declare/3.

If argument 'x-expires' is given to arguments, then it will be used instead of queue_ttl.

If argument 'x-max-priority' is given to arguments, then it will be used instead of queue_max_priority.

deadletter_exchange - name or {type, name} of the deadletter exchange (Default: Same as exchange name suffixed by .deadletter). If it does not exist, it will be created. For valid exchange types see GenRMQ.Binding

deadletter_routing_key - defines name of the deadletter routing key (Default: #).

Examples:

def init() do
  [
    connection: "amqp://guest:guest@localhost:5672",
    queue: "gen_rmq_in_queue",
    queue_options: [
      durable: true,
      passive: true,
      arguments: [
        {"x-queue-type", :longstr ,"quorum"}
      ]
    ]
    exchange: "gen_rmq_exchange",
    routing_key: "#",
    prefetch_count: "10",
    uri: "amqp://guest:guest@localhost:5672",
    concurrency: true,
    terminate_timeout: 5_000,
    handle_message_timeout: 5_000,
    queue_ttl: 5_000,
    retry_delay_function: fn attempt -> :timer.sleep(1000 * attempt) end,
    reconnect: true,
    deadletter: true,
    deadletter_queue: "gen_rmq_in_queue_error",
    deadletter_queue_options: [
      arguments: [
        {"x-queue-type", :longstr ,"quorum"}
      ]
    ]
    deadletter_exchange: "gen_rmq_exchange.deadletter",
    deadletter_routing_key: "#",
    queue_max_priority: 10
  ]
end