EctoPGMQ (ecto_pgmq v2.0.0)

Copy Markdown View Source

An opinionated PGMQ client for Elixir that builds on top of Ecto and the Ecto.Adapters.Postgres adapter.

Summary

Types

delay()

A delay before a message becomes visible.

destination()

A message destination.

message_update_attributes()

Message update attributes.

notification_throttle()

The minimum time between notifications.

partition_config()

A queue partition configuration.

poll_config()

A message polling configuration.

queue_create_attributes()

Queue creation attributes.

queue_update_attributes()

Queue update attributes.

read_messages_opts()

Options for reading messages.

visibility_timeout()

The time from now that a message is invisible.

Message API

archive_messages(repo, queue, message_ids, opts \\ [])

Archives the given messages from the given queue.

delete_messages(repo, queue, message_ids, opts \\ [])

Deletes the given messages from the given queue.

read_messages(repo, queue, visibility_timeout, quantity, opts \\ [])

Reads messages from the given queue.

send_messages(repo, destination, messages, opts \\ [])

Sends messages to the given queue.

update_messages(repo, queue, message_ids, map, opts \\ [])

Updates the given messages in the given queue.

Queue API

all_queues(repo, opts \\ [])

Lists all queues.

create_queue(repo, queue, attributes \\ %{}, opts \\ [])

Creates a queue with the given name.

drop_queue(repo, queue, opts \\ [])

Drops the given queue.

get_queue(repo, queue, opts \\ [])

Gets the given queue.

purge_queue(repo, queue, opts \\ [])

Purges the given queue.

update_queue(repo, queue, attributes, opts \\ [])

Updates the given queue.

Types

delay()

@type delay() :: Duration.t() | EctoPGMQ.PGMQ.delay()

A delay before a message becomes visible.

This can take any of the following forms:

  • A Duration.t/0 denoting the time to wait before a message becomes visible.

  • An integer/0 denoting the time (in seconds) to wait before a message becomes visible.

  • A DateTime.t/0 denoting when a message should become visible.

For more information about this type, see EctoPGMQ.PGMQ.delay/0.

destination()

@type destination() ::
  EctoPGMQ.Queue.name()
  | {:queue, EctoPGMQ.Queue.name()}
  | {:routing_key, EctoPGMQ.PGMQ.routing_key()}

A message destination.

message_update_attributes()

@type message_update_attributes() :: %{visibility_timeout: delay()}

Message update attributes.

The following attributes are supported:

  • :visibility_timeout - A required delay/0 for the messages.

notification_throttle()

@type notification_throttle() :: Duration.t() | EctoPGMQ.PGMQ.throttle_interval()

The minimum time between notifications.

This can take either of the following forms:

  • A Duration.t/0 denoting the minimum time between notifications.

  • A non_neg_integer/0 denoting the minimum time (in milliseconds) between notifications.

For more information about notifications, see EctoPGMQ.Notifications.

For more information about this type, see EctoPGMQ.PGMQ.throttle_interval/0.

partition_config()

@type partition_config() ::
  {EctoPGMQ.PGMQ.partition_interval(), EctoPGMQ.PGMQ.retention_interval()}

A queue partition configuration.

A partition configuration is a tuple containing two elements: the partition interval and the retention interval.

Both elements can take either of the following forms:

For more information about partitioning, see Partitioning.

For more information about this type, see EctoPGMQ.PGMQ.partition_interval/0 and EctoPGMQ.PGMQ.retention_interval/0.

poll_config()

@type poll_config() ::
  {Duration.t() | EctoPGMQ.PGMQ.poll_interval(),
   Duration.t() | EctoPGMQ.PGMQ.poll_timeout()}

A message polling configuration.

A polling configuration is a tuple containing two elements: the poll interval and the poll timeout.

Both elements can take either of the following forms:

  • A Duration.t/0 denoting a length of time.

  • A pos_integer/0 denoting a length of time. The unit for the poll interval is milliseconds and the unit for the timeout is seconds.

For more information about polling, see Polling.

For more information about this type, see EctoPGMQ.PGMQ.poll_interval/0 and EctoPGMQ.PGMQ.poll_timeout/0.

queue_create_attributes()

@type queue_create_attributes() :: %{
  optional(:bindings) => [EctoPGMQ.Binding.pattern()],
  optional(:message_groups?) => boolean(),
  optional(:notifications) => notification_throttle() | nil,
  optional(:partitions) => partition_config() | nil,
  optional(:unlogged?) => boolean()
}

Queue creation attributes.

The following attributes are supported:

  • :bindings - An optional list/0 of EctoPGMQ.Binding.pattern/0 for the queue. Defaults to []. For more information about bindings, see Message Routing.

  • :message_groups? - An optional boolean/0 denoting whether or not the queue should be optimized for FIFO message group reads. Defaults to false. For more information about FIFO message groups, see FIFO Message Groups.

  • :notifications - An optional notification_throttle/0 for the queue or nil to leave notifications disabled. Defaults to nil. For more information about notifications, see EctoPGMQ.Notifications.

  • :partitions - An optional partition_config/0 for the queue or nil to disable partitioning. This option is ignored for unlogged queues. Defaults to nil. For more information about partitioning, see Partitioning.

  • :unlogged? - An optional boolean/0 denoting whether or not the queue should be unlogged. Defaults to false.

queue_update_attributes()

@type queue_update_attributes() :: %{
  optional(:bindings) => [EctoPGMQ.Binding.pattern()],
  optional(:message_groups?) => true,
  optional(:notifications) => notification_throttle() | nil
}

Queue update attributes.

The following attributes are supported:

  • :bindings - An optional list/0 of EctoPGMQ.Binding.pattern/0 for the queue. For more information about bindings, see Message Routing.

    Replace Behavior

    When given, this attribute will REPLACE the existing bindings for the queue. This means that this attribute must contain ALL of the desired bindings for the queue, not just a delta.

  • :message_groups? - true to optimize the queue for FIFO message group reads. Note that true is the only valid value because this operation cannot be undone. For more information about FIFO message groups, see FIFO Message Groups.

  • :notifications - An optional notification_throttle/0 for the queue or nil to disable notifications. For more information about notifications, see EctoPGMQ.Notifications.

read_messages_opts()

@type read_messages_opts() :: [
  {:delete?, boolean()}
  | {:polling, poll_config() | nil}
  | {:payload_type, EctoPGMQ.Message.payload_type()}
  | {:message_grouping, :head | :round_robin | :throughput_optimized | nil}
  | EctoPGMQ.PGMQ.query_opt()
]

Options for reading messages.

In addition to the standard query options, messages can be read with the following options:

  • :delete? - An optional boolean/0 denoting whether or not to delete messages immediately after reading them. Defaults to false. For more information, see EctoPGMQ.PGMQ.pop/4.

  • :message_grouping - An optional value specifying how to handle message groups when reading messages. Possible values are :round_robin, :throughput_optimized, or nil to ignore message groups when reading. This option is ignored when deleting on read. Defaults to nil. For more information about FIFO message groups, see FIFO Message Groups.

  • :payload_type - An optional EctoPGMQ.Message.payload_type/0 for the message payloads. Defaults to :map.

  • :polling - An optional poll_config/0 for the read operation or nil to disable polling. This option is ignored when deleting on read. Defaults to nil. For more information about polling, see Polling.

visibility_timeout()

@type visibility_timeout() :: Duration.t() | EctoPGMQ.PGMQ.visibility_timeout()

The time from now that a message is invisible.

This can take either of the following forms:

  • A Duration.t/0 denoting how long a message is invisible.

  • An integer/0 denoting how long (in seconds) a message is invisible.

For more information about this type, see EctoPGMQ.PGMQ.visibility_timeout/0.

Message API

archive_messages(repo, queue, message_ids, opts \\ [])

@spec archive_messages(
  Ecto.Repo.t(),
  EctoPGMQ.Queue.name(),
  [EctoPGMQ.Message.id()],
  [
    EctoPGMQ.PGMQ.query_opt()
  ]
) :: :ok

Archives the given messages from the given queue.

Options

This function supports the standard query options.

Examples

iex> messages = [Message.build(%{"id" => 1})]
iex> %{"my_queue" => ids} = EctoPGMQ.send_messages(Repo, "my_queue", messages)
iex> EctoPGMQ.archive_messages(Repo, "my_queue", ids)
:ok

delete_messages(repo, queue, message_ids, opts \\ [])

@spec delete_messages(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [EctoPGMQ.Message.id()], [
  EctoPGMQ.PGMQ.query_opt()
]) :: :ok

Deletes the given messages from the given queue.

Options

This function supports the standard query options.

Examples

iex> messages = [Message.build(%{"id" => 1})]
iex> %{"my_queue" => ids} = EctoPGMQ.send_messages(Repo, "my_queue", messages)
iex> EctoPGMQ.delete_messages(Repo, "my_queue", ids)
:ok

read_messages(repo, queue, visibility_timeout, quantity, opts \\ [])

@spec read_messages(
  Ecto.Repo.t(),
  EctoPGMQ.Queue.name(),
  visibility_timeout(),
  EctoPGMQ.PGMQ.quantity(),
  read_messages_opts()
) :: [EctoPGMQ.Message.t()]

Reads messages from the given queue.

Options

See read_messages_opts/0 for information about the options supported by this function.

Examples

iex> vt = Duration.new!(second: 5)
iex> messages = [Message.build(%{"id" => 1})]
iex> EctoPGMQ.send_messages(Repo, "my_queue", messages)
iex> [%Message{reads: 1}] = EctoPGMQ.read_messages(Repo, "my_queue", vt, 2)

send_messages(repo, destination, messages, opts \\ [])

@spec send_messages(Ecto.Repo.t(), destination(), [EctoPGMQ.Message.message()], [
  {:delay, delay()}
  | {:payload_type, EctoPGMQ.Message.payload_type()}
  | EctoPGMQ.PGMQ.query_opt()
]) :: EctoPGMQ.PGMQ.queue_message_ids()

Sends messages to the given queue.

Options

In addition to the standard query options, this function also supports the following options:

Examples

iex> messages = [Message.build(%{"id" => 1})]
iex> delay = Duration.new!(hour: 1)
iex> %{"my_queue" => [id]} = EctoPGMQ.send_messages(Repo, "my_queue", messages, delay: delay)
iex> is_integer(id)
true

iex> messages = [Message.build(%{"id" => 1})]
iex> EctoPGMQ.PGMQ.bind_topic(Repo, "#", "my_queue")
iex> destination = {:routing_key, "my.routing.key"}
iex> %{"my_queue" => [id]} = EctoPGMQ.send_messages(Repo, destination, messages)
iex> is_integer(id)
true

update_messages(repo, queue, message_ids, map, opts \\ [])

@spec update_messages(
  Ecto.Repo.t(),
  EctoPGMQ.Queue.name(),
  [EctoPGMQ.Message.id()],
  message_update_attributes(),
  [EctoPGMQ.PGMQ.query_opt()]
) :: :ok

Updates the given messages in the given queue.

Options

This function supports the standard query options.

Examples

iex> messages = [Message.build(%{"id" => 1})]
iex> visibility_timeout = Duration.new!(minute: 5)
iex> %{"my_queue" => ids} = EctoPGMQ.send_messages(Repo, "my_queue", messages)
iex> EctoPGMQ.update_messages(Repo, "my_queue", ids, %{visibility_timeout: visibility_timeout})
:ok

Queue API

all_queues(repo, opts \\ [])

@spec all_queues(Ecto.Repo.t(), [EctoPGMQ.PGMQ.query_opt()]) :: [EctoPGMQ.Queue.t()]

Lists all queues.

Options

This function supports the standard query options.

Examples

iex> [%Queue{} | _] = EctoPGMQ.all_queues(Repo)

create_queue(repo, queue, attributes \\ %{}, opts \\ [])

@spec create_queue(Ecto.Repo.t(), EctoPGMQ.Queue.name(), queue_create_attributes(), [
  EctoPGMQ.PGMQ.query_opt()
]) :: EctoPGMQ.Queue.t()

Creates a queue with the given name.

To create a queue in an Ecto.Migration, see EctoPGMQ.Migrations.create_queue/2.

Options

This function supports the standard query options.

Examples

iex> queue = EctoPGMQ.create_queue(Repo, "my_unpartitioned_queue", %{notifications: 1_000})
iex> %Queue{notifications: %Throttle{}} = queue

iex> queue = EctoPGMQ.create_queue(Repo, "my_partitioned_queue", %{partitions: {10_000, 100_000}})
iex> %Queue{partitioned?: true} = queue

iex> partitions = {Duration.new!(hour: 1), Duration.new!(day: 1)}
iex> queue = EctoPGMQ.create_queue(Repo, "my_partitioned_queue", %{partitions: partitions})
iex> %Queue{partitioned?: true} = queue

iex> queue = EctoPGMQ.create_queue(Repo, "my_unlogged_queue", %{unlogged?: true})
iex> %Queue{unlogged?: true} = queue

drop_queue(repo, queue, opts \\ [])

@spec drop_queue(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [EctoPGMQ.PGMQ.query_opt()]) ::
  :ok

Drops the given queue.

To drop a queue in an Ecto.Migration, see EctoPGMQ.Migrations.drop_queue/2.

Options

This function supports the standard query options.

Examples

iex> EctoPGMQ.drop_queue(Repo, "my_queue")
:ok

get_queue(repo, queue, opts \\ [])

@spec get_queue(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [EctoPGMQ.PGMQ.query_opt()]) ::
  EctoPGMQ.Queue.t() | nil

Gets the given queue.

Options

This function supports the standard query options.

Examples

iex> %Queue{} = EctoPGMQ.get_queue(Repo, "my_queue")

iex> EctoPGMQ.get_queue(Repo, "my_non_existent_queue")
nil

purge_queue(repo, queue, opts \\ [])

@spec purge_queue(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [EctoPGMQ.PGMQ.query_opt()]) ::
  EctoPGMQ.PGMQ.purged_messages()

Purges the given queue.

Options

This function supports the standard query options.

Examples

iex> messages = [Message.build(%{"id" => 1})]
iex> EctoPGMQ.send_messages(Repo, "my_queue", messages)
iex> EctoPGMQ.purge_queue(Repo, "my_queue")
1

update_queue(repo, queue, attributes, opts \\ [])

@spec update_queue(Ecto.Repo.t(), EctoPGMQ.Queue.name(), queue_update_attributes(), [
  EctoPGMQ.PGMQ.query_opt()
]) :: EctoPGMQ.Queue.t()

Updates the given queue.

Unexpected Results

Because the underlying tables are owned by PGMQ, this function avoids row locks so as not to disturb any internal PGMQ processes. As a result, if multiple processes attempt to update a queue simultaneously, they may get unexpected results.

To update a queue in an Ecto.Migration, see EctoPGMQ.Migrations.update_queue/3.

Options

This function supports the standard query options.

Examples

iex> throttle = Duration.new!(second: 5)
iex> queue = EctoPGMQ.update_queue(Repo, "my_queue", %{notifications: throttle})
iex> %Queue{notifications: %Throttle{}} = queue