View Source Lapin.Consumer behaviour (lapin v2.0.0)

Extensible behaviour to define consumer configuration.

Lapin provides a number of submodules which implement the patterns found in the RabbitMQ Tutorials.

defmodule ExampleApp.SomeConsumer do
  use Lapin.Consumer

  [... callbacks implementation ...]
end

Summary

Types

config()

Consumer configuration

consumer_tag()

Consumer Tag

prefetch_count()

Consumer Prefetch

t()

Lapin Consumer Behaviour

Callbacks

ack(consumer)

Consumer acknowledgements enabled

prefetch_count(consumer)

Consumer message prefetch count

queue(consumer)

Queue to consume from

Functions

ack_message(map, delivery_tag)

ACK message consumption

create(connection, config)

Creates a consumer from configuration

get(consumers, consumer_tag)

Find consumer by consumer_tag

reject_message(map, delivery_tag, requeue)

Reject message

Types

config()

@type config() :: Keyword.t()

Consumer configuration

The following keys are supported:

pattern: producer pattern (module using the Lapin.Producer behaviour)

If using the Lapin.Consumer.Config default implementation, the following keys are also supported:

queue: queue to consume from, (String.t(), required)
ack: producer ack (boolean(), default: false*
prefetch_count: consumer prefetch count (integer(), default: 1)

consumer_tag()

@type consumer_tag() :: String.t()

Consumer Tag

prefetch_count()

@type prefetch_count() :: integer()

Consumer Prefetch

t()

@type t() :: %Lapin.Consumer{
  channel: AMQP.Channel.t(),
  config: config(),
  consumer_tag: consumer_tag(),
  pattern: atom(),
  queue: String.t()
}

Lapin Consumer Behaviour

Callbacks

ack(consumer)

@callback ack(consumer :: t()) :: boolean()

Consumer acknowledgements enabled

prefetch_count(consumer)

@callback prefetch_count(consumer :: t()) :: prefetch_count()

Consumer message prefetch count

queue(consumer)

@callback queue(consumer :: t()) :: Lapin.Queue.t()

Queue to consume from

Functions

ack_message(map, delivery_tag)

@spec ack_message(t(), integer()) :: :ok | {:error, term()}

ACK message consumption

create(connection, config)

@spec create(AMQP.Connection.t(), config()) :: t()

Creates a consumer from configuration

get(consumers, consumer_tag)

@spec get([t()], consumer_tag()) :: {:ok, t()} | {:error, :not_found}

Find consumer by consumer_tag

reject_message(map, delivery_tag, requeue)

@spec reject_message(t(), integer(), boolean()) :: :ok | {:error, term()}

Reject message