View Source AMQPHelpers.Reliability.Producer (AMQP Helpers v1.5.0)

A producer for dealing with reliability scenarios.

A AMQPHelpers.Reliability.Producer is process which manages a reliable send operation over AMQP, where messages are acknowledge by the broker after sending them. Publisher Confirms AMQP extension is used to achieve this.

Example

This Producer enforces the usage of AMQP.Application, so after defining our application connection and channel we can start a producer and start publishing message in a reliable way.

alias AMQPHelpers.Reliability.Producer

{:ok, producer} = Producer.start_link(channel_name: :my_channel_name)

Producer.publish(producer, "exchange", "routing_key", "payload", message_id: "foo")

Summary

Types

option()

Option values used by start_link/1 function.

options()

Options used by start_link/1 function.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

publish(server, exchange, routing_key, payload, opts, timeout \\ 5000)

Publish a message ensuring that the broker receives it.

setup_channel(producer)

Setups the channel for reliable message delivery.

start_link(opts \\ [])

Starts a Producer process linked to the current process.

Types

option()

@type option() ::
  GenServer.option()
  | {:adapter, module()}
  | {:setup_channel_on_init, boolean()}
  | {:channel, AMQP.Channel.t()}
  | {:channel_name, binary() | atom()}
  | {:retry_interval, non_neg_integer()}

Option values used by start_link/1 function.

options()

@type options() :: [option()]

Options used by start_link/1 function.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

publish(server, exchange, routing_key, payload, opts, timeout \\ 5000)

@spec publish(
  GenServer.server(),
  AMQP.Basic.exchange(),
  AMQP.Basic.routing_key(),
  AMQP.Basic.payload(),
  keyword(),
  timeout()
) :: :ok | {:error, term()}

Publish a message ensuring that the broker receives it.

Similar to AMQPHelpers.Adapter.publish/5 but ensures that the broker receives the message sent. Some message delivery options are enabled by default (mandatory and persistent) and message_id option is required to track the message state.

setup_channel(producer)

@spec setup_channel(GenServer.server()) :: :ok

Setups the channel for reliable message delivery.

This function is used to setup reliable message delivery when setup_channel_on_init is set to false. Not required by default but useful for testing purposes.

start_link(opts \\ [])

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

Starts a Producer process linked to the current process.

Options

The following option can be given to Producer when starting it. Note that queue_name is required.

adapter - Sets the AMQPHelpers.Adapter. Defaults to AMQPHelpers.Adapters.AMQP.
setup_channel_on_init - Whether to configure the channel for reliable message sending on init or not. Defaults to false.
channel - The channel to use to consume messages. NOTE: do not use this for production environments because this producer does not supervise the given channel. Instead, use channel_name which makes use of AMQP.Application.
channel_name - The name of the configured channel to use. See AMQP.Application for more information. Defaults to :default.
retry_interval - The number of millisecond to wait if an error happens when trying to consume messages or when trying to open a channel.

GenServer.options/0 are also available. See GenServer.start_link/2 for more information about these.