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

RabbitMQ connection handler

This module handles the RabbitMQ connection. It also provides a behaviour for worker module implementation. The worker module should use the Lapin.Connection behaviour and implement the callbacks it needs.

When using the Lapin.Connection behaviour a publish/4 function is injected in the worker module as a shortcut to the Lapin.Connection.publish/5 function which removes the need for passing in the connection and is publicly callable to publish messages on the connection configured for the implementing module.

Summary

Types

config()

Connection configuration

on_callback()

Callback result

on_deliver()

handle_deliver/2 callback result

reason()

Reason for message rejection

t()

Connection

Callbacks

handle_cancel(t)

Called when receiving a basic.cancel from the broker.

handle_cancel_ok(t)

Called when receiving a basic.cancel_ok from the broker.

handle_consume_ok(t)

Called when receiving a basic.consume_ok from the broker.

handle_deliver(t, t)

Called when receiving a basic.deliver from the broker.

handle_publish(t, t)

Called when completing a basic.publish with the broker.

handle_return(t, t)

Called when receiving a basic.return from the broker.

payload_for(t, t)

Called before handle_deliver/2 to get the payload type.

Functions

close(connection)

Closes the connection

publish(connection, exchange, routing_key, payload, options \\ [])

Publishes a message to the specified exchange with the given routing_key

start_link(configuration, options \\ [])

Starts a Lapin.Connection with the specified configuration

Types

config()

@type config() :: [
  consumers: [Lapin.Consumer.config()],
  producers: [Lapin.Producer.config()]
]

Connection configuration

The following keys are supported:

module: module using the Lapin.Connection behaviour
uri: AMQP URI (String.t | URI.t)
host: broker hostname (string | charlist), default: 'localhost'
port: broker port (string | integer), default: 5672
virtual_host: broker vhost (string), default: "/"
username: username (string)
password: password (string)
auth_mechanisms: broker auth_mechanisms ([:amqplain | :external | :plain]), default: amqp_client default
ssl_options: ssl options ([:ssl:ssl_option]), default: none
producers: producers to configure ([Producer.config]), default: []
consumers: consumers to configure ([Consumer.config]), default: []

on_callback()

@type on_callback() :: :ok | {:error, message :: String.t()}

Callback result

on_deliver()

@type on_deliver() :: :ok | {:reject, reason()} | term()

handle_deliver/2 callback result

reason()

@type reason() :: term()

Reason for message rejection

t()

@type t() :: :gen_statem.server_ref()

Connection

Callbacks

handle_cancel(t)

@callback handle_cancel(Lapin.Consumer.t()) :: on_callback()

Called when receiving a basic.cancel from the broker.

handle_cancel_ok(t)

@callback handle_cancel_ok(Lapin.Consumer.t()) :: on_callback()

Called when receiving a basic.cancel_ok from the broker.

handle_consume_ok(t)

@callback handle_consume_ok(Lapin.Consumer.t()) :: on_callback()

Called when receiving a basic.consume_ok from the broker.

This signals successul registration as a consumer.

handle_deliver(t, t)

@callback handle_deliver(Lapin.Consumer.t(), Lapin.Message.t()) :: on_deliver()

Called when receiving a basic.deliver from the broker.

Return values from this callback determine message acknowledgement:

:ok: Message was processed by the consumer and should be removed from queue
{:reject, reason}: Message was not processed and should be rejected

Any other return value requeues the message to prevent data loss. A crash in the callback code will however reject the message to prevent loops if the message was already delivered before.

The reason term can be used by the application to signal the reason of rejection and is logged in debug.

handle_publish(t, t)

@callback handle_publish(Lapin.Producer.t(), Lapin.Message.t()) :: on_callback()

Called when completing a basic.publish with the broker.

Message transmission to the broker is successful when this callback is called.

handle_return(t, t)

@callback handle_return(Lapin.Consumer.t(), Lapin.Message.t()) :: on_callback()

Called when receiving a basic.return from the broker.

This signals an undeliverable returned message from the broker.

payload_for(t, t)

@callback payload_for(Lapin.Consumer.t(), Lapin.Message.t()) :: Lapin.Message.Payload.t()

Called before handle_deliver/2 to get the payload type.

Should return a data type instance to decode the payload into. A Lapin.Message.Payload implementation must be provided for this type. The default implementation leaves the payload unaltered.

Functions

close(connection)

@spec close(connection :: t()) :: on_callback()

Closes the connection

publish(connection, exchange, routing_key, payload, options \\ [])

@spec publish(
  connection :: t(),
  String.t(),
  String.t(),
  Lapin.Message.Payload.t(),
  options :: Keyword.t()
) :: on_callback()

Publishes a message to the specified exchange with the given routing_key

start_link(configuration, options \\ [])

@spec start_link(config(), options :: GenServer.options()) :: GenServer.on_start()

Starts a Lapin.Connection with the specified configuration