An SDK that fully covers the PGMQ API-space.
Instead of implementing a corresponding function for every alias and parameterization supported by PGMQ, this module relies on client-side defaults to implement a single function for each distinct piece of PGMQ functionality.
Partitioning
PGMQ supports partitioning both queues and archives.
The pg_partman extension must be available in order to use partitioning.
For more information about partitioning, see the PGMQ docs.
Polling
PGMQ supports Postgres server-side polling during read operations. Reading with a poll can be used to reduce network round trips if there is a good chance that demand can be satisfied in a short time BUT doing so utilizes a connection for the duration of the read operation. As such, polling should be avoided in situations where the DB connection pool is a bottleneck.
Query Options
All of the functions in this module support a common set of query options.
For a detailed description of these options, see query_opt/0.
Summary
Types
Filter conditions to be applied when reading messages from a queue.
A delay before a message becomes visible.
The number of partitions to create preemptively.
The interval at which new partitions should be created.
A message payload.
The time (in milliseconds) to wait between polls.
The maximum time (in seconds) to poll for messages.
The number of purged messages.
The maximum number of messages to read.
A query configuration option.
A map of queues and the IDs of the messages that were sent to them.
The interval at which old partitions should be dropped.
A PGMQ routing key.
The minimum time (in milliseconds) between notifications.
The time from now (in seconds) that a message is invisible.
Functions
Archives the given messages in the given queue.
Binds the given queue to the given pattern.
Converts the archive for the given queue into a partitioned table.
Creates an index to optimize FIFO message group read performance for the given queue.
Creates an unpartitioned queue with the given name.
Creates a partitioned queue with the given name.
Creates an unlogged queue with the given name.
Deletes the given messages from the given queue.
Disables insert notifications for the given queue.
Drops the given queue.
Enables insert notifications for the given queue.
Lists all insert notification throttles.
Lists all queues.
Lists all topic bindings.
Returns metrics for all queues.
Simultaneously fetches and deletes messages from the given queue.
Purges all messages from the given queue and returns the number of messages that were deleted.
Reads messages from the given queue.
Reads messages from the given queue while respecting FIFO message groups and optimizing throughput.
Reads messages from the given queue while respecting FIFO message groups and returning a single message per group.
Reads messages from the given queue with a Postgres server-side poll while respecting FIFO message groups and returning a single message per group.
Reads messages from the given queue while respecting FIFO message groups and round-robin interleaving FIFO message groups.
Reads messages from the given queue with a Postgres server-side poll while respecting FIFO message groups and round-robin interleaving FIFO message groups.
Reads messages from the given queue with a Postgres server-side poll while respecting FIFO message groups and optimizing throughput.
Reads messages from the given queue with a Postgres server-side poll.
Sends the given messages to the given queue.
Sends the given messages with the given routing key.
Sets the visibility timeout of the given messages in the given queue.
Returns all of the bindings that match the given routing key.
Unbinds the given queue from the given pattern.
Updates the insert notification throttle for the given queue.
Validates the given routing key.
Validates the given binding pattern.
Types
Filter conditions to be applied when reading messages from a queue.
Note that the filter conditions are applied to the message body, not the headers.
Experimental Feature
As stated in the PGMQ docs, conditional message reading is an experimental feature and the API might be subject to change in future releases.
@type delay() :: non_neg_integer() | DateTime.t()
A delay before a message becomes visible.
This can take either of the following forms:
An
integer/0denoting the time (in seconds) to wait before a message becomes visible.A
DateTime.t/0denoting when a message should become visible.
@type leading_partitions() :: non_neg_integer()
The number of partitions to create preemptively.
@type partition_interval() :: pos_integer() | Duration.t()
The interval at which new partitions should be created.
This can take either of the following forms:
A
pos_integer/0denoting how many messages per partition.A
Duration.t/0denoting a time range per partition.
@type payload() :: %{optional(String.Chars.t()) => term()}
A message payload.
@type poll_interval() :: pos_integer()
The time (in milliseconds) to wait between polls.
@type poll_timeout() :: pos_integer()
The maximum time (in seconds) to poll for messages.
@type purged_messages() :: non_neg_integer()
The number of purged messages.
@type quantity() :: pos_integer()
The maximum number of messages to read.
A query configuration option.
The following query configuration options are supported:
@type queue_message_ids() :: %{ optional(EctoPGMQ.Queue.name()) => [EctoPGMQ.Message.id()] }
A map of queues and the IDs of the messages that were sent to them.
@type retention_interval() :: pos_integer() | Duration.t()
The interval at which old partitions should be dropped.
This can take either of the following forms:
A
pos_integer/0denoting how many messages to retain in total.A
Duration.t/0denoting a total time range to retain.
@type routing_key() :: String.t()
A PGMQ routing key.
Routing keys must meet the following conditions:
Must only contain alphanumeric characters, dots (
.), hyphens (-), and underscores (_).Cannot start with a dot.
Cannot contain consecutive dots.
Cannot be longer than 255 characters.
Routing keys can be validated with validate_routing_key/3.
For more information about message routing, see Message Routing.
@type throttle_interval() :: non_neg_integer()
The minimum time (in milliseconds) between notifications.
A throttle interval of 0 effectively disables notification throttling.
For more information about notification throttling, see Throttling.
@type visibility_timeout() :: integer()
The time from now (in seconds) that a message is invisible.
Functions
@spec archive(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [EctoPGMQ.Message.id()], [ query_opt() ]) :: :ok
Archives the given messages in the given queue.
For more information about this function, see the PGMQ docs.
Examples
iex> message_ids = PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}, %{"id" => 2}])
iex> PGMQ.archive(Repo, "my_queue", message_ids)
:ok@spec bind_topic(Ecto.Repo.t(), EctoPGMQ.Binding.pattern(), EctoPGMQ.Queue.name(), [ query_opt() ]) :: :ok
Binds the given queue to the given pattern.
For more information about message routing, see Message Routing.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.bind_topic(Repo, "#", "my_queue")
:ok@spec convert_archive_partitioned( Ecto.Repo.t(), EctoPGMQ.Queue.name(), partition_interval(), retention_interval(), leading_partitions(), [query_opt()] ) :: :ok
Converts the archive for the given queue into a partitioned table.
Previous Archive
This function postfixes the old archive table name with _old and leaves
its contents untouched. Additional cleanup (table deletion, message
movement, etc.) is left to the user.
For more information about partitioning, see Partitioning.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.convert_archive_partitioned(Repo, "my_queue", 10_000, 100_000, 10)
:ok
iex> partition = Duration.new!(hour: 1)
iex> retention = Duration.new!(day: 1)
iex> PGMQ.convert_archive_partitioned(Repo, "my_queue", partition, retention, 10)
:ok@spec create_fifo_index(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [query_opt()]) :: :ok
Creates an index to optimize FIFO message group read performance for the given queue.
For more information about FIFO message groups, see FIFO Message Groups.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.create_fifo_index(Repo, "my_queue")
:ok@spec create_non_partitioned(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [query_opt()]) :: :ok
Creates an unpartitioned queue with the given name.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.create_non_partitioned(Repo, "my_unpartitioned_queue")
:ok@spec create_partitioned( Ecto.Repo.t(), EctoPGMQ.Queue.name(), partition_interval(), retention_interval(), [query_opt()] ) :: :ok
Creates a partitioned queue with the given name.
For more information about partitioning, see Partitioning.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.create_partitioned(Repo, "my_partitioned_queue", 10_000, 100_000)
:ok
iex> partition = Duration.new!(hour: 1)
iex> retention = Duration.new!(day: 1)
iex> PGMQ.create_partitioned(Repo, "my_partitioned_queue", partition, retention)
:ok@spec create_unlogged(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [query_opt()]) :: :ok
Creates an unlogged queue with the given name.
For more information about this function, see the PGMQ docs.
Unlogged Tables
Unlogged tables benefit from faster write operations but they risk data loss if the Postgres server restarts. Use with caution.
Examples
iex> PGMQ.create_unlogged(Repo, "my_unlogged_queue")
:ok@spec delete(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [EctoPGMQ.Message.id()], [ query_opt() ]) :: :ok
Deletes the given messages from the given queue.
For more information about this function, see the PGMQ docs.
Examples
iex> message_ids = PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}, %{"id" => 2}])
iex> PGMQ.delete(Repo, "my_queue", message_ids)
:ok@spec disable_notify_insert(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [query_opt()]) :: :ok
Disables insert notifications for the given queue.
For more information about notifications, see EctoPGMQ.Notifications.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.enable_notify_insert(Repo, "my_queue")
iex> PGMQ.disable_notify_insert(Repo, "my_queue")
:ok@spec drop_queue(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [query_opt()]) :: :ok
Drops the given queue.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.drop_queue(Repo, "my_queue")
:ok@spec enable_notify_insert( Ecto.Repo.t(), EctoPGMQ.Queue.name(), throttle_interval(), [query_opt()] ) :: :ok
Enables insert notifications for the given queue.
For more information about notifications, see EctoPGMQ.Notifications.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.enable_notify_insert(Repo, "my_queue", 1_000)
:ok@spec list_notify_insert_throttles(Ecto.Repo.t(), [query_opt()]) :: [ EctoPGMQ.Throttle.t() ]
Lists all insert notification throttles.
For more information about notifications, see EctoPGMQ.Notifications.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.enable_notify_insert(Repo, "my_queue")
iex> [%Throttle{} | _] = PGMQ.list_notify_insert_throttles(Repo)@spec list_queues(Ecto.Repo.t(), [query_opt()]) :: [EctoPGMQ.Queue.t()]
Lists all queues.
Because this function naively wraps the corresponding PGMQ function, the
:bindings, :metrics and :notifications fields in the returned
EctoPGMQ.Queue structs will not be populated.
For more information about this function, see the PGMQ docs.
Examples
iex> [%Queue{} | _] = PGMQ.list_queues(Repo)@spec list_topic_bindings(Ecto.Repo.t(), [query_opt()]) :: [EctoPGMQ.Binding.t()]
Lists all topic bindings.
For more information about message routing, see Message Routing.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.bind_topic(Repo, "#", "my_queue")
iex> [%Binding{} | _] = PGMQ.list_topic_bindings(Repo)@spec metrics_all(Ecto.Repo.t(), [query_opt()]) :: [EctoPGMQ.Metrics.t()]
Returns metrics for all queues.
For more information about this function, see the PGMQ docs.
Examples
iex> [%Metrics{} | _] = PGMQ.metrics_all(Repo)@spec pop(Ecto.Repo.t(), EctoPGMQ.Queue.name(), quantity(), [query_opt()]) :: [ EctoPGMQ.Message.t() ]
Simultaneously fetches and deletes messages from the given queue.
This function does NOT increment the read count of the fetched messages.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 0}] = PGMQ.pop(Repo, "my_queue", 1)@spec purge_queue(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [query_opt()]) :: purged_messages()
Purges all messages from the given queue and returns the number of messages that were deleted.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> PGMQ.purge_queue(Repo, "my_queue")
1@spec read( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), conditional(), [ query_opt() ] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read(Repo, "my_queue", 5, 1)@spec read_grouped( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), [query_opt()] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue while respecting FIFO message groups and optimizing throughput.
For more information about FIFO message groups, see FIFO Message Groups.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read_grouped(Repo, "my_queue", 5, 1)@spec read_grouped_head( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), [ query_opt() ] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue while respecting FIFO message groups and returning a single message per group.
For more information about FIFO message groups, see FIFO Message Groups.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read_grouped_head(Repo, "my_queue", 5, 1)@spec read_grouped_head_with_poll( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), poll_timeout(), poll_interval(), [query_opt()] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue with a Postgres server-side poll while respecting FIFO message groups and returning a single message per group.
For more information about FIFO message groups, see FIFO Message Groups.
For more information about polling, see Polling.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read_grouped_head_with_poll(Repo, "my_queue", 5, 1, 5, 500)@spec read_grouped_rr( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), [ query_opt() ] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue while respecting FIFO message groups and round-robin interleaving FIFO message groups.
For more information about FIFO message groups, see FIFO Message Groups.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read_grouped_rr(Repo, "my_queue", 5, 1)@spec read_grouped_rr_with_poll( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), poll_timeout(), poll_interval(), [query_opt()] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue with a Postgres server-side poll while respecting FIFO message groups and round-robin interleaving FIFO message groups.
For more information about FIFO message groups, see FIFO Message Groups.
For more information about polling, see Polling.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read_grouped_rr_with_poll(Repo, "my_queue", 5, 1, 5, 500)@spec read_grouped_with_poll( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), poll_timeout(), poll_interval(), [query_opt()] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue with a Postgres server-side poll while respecting FIFO message groups and optimizing throughput.
For more information about FIFO message groups, see FIFO Message Groups.
For more information about polling, see Polling.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read_grouped_with_poll(Repo, "my_queue", 5, 1, 5, 500)@spec read_with_poll( Ecto.Repo.t(), EctoPGMQ.Queue.name(), visibility_timeout(), quantity(), poll_timeout(), poll_interval(), conditional(), [query_opt()] ) :: [EctoPGMQ.Message.t()]
Reads messages from the given queue with a Postgres server-side poll.
For more information about polling, see Polling.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [%Message{reads: 1}] = PGMQ.read_with_poll(Repo, "my_queue", 5, 1, 5, 500)@spec send_batch( Ecto.Repo.t(), EctoPGMQ.Queue.name(), [payload() | nil], [EctoPGMQ.Message.headers() | nil] | nil, delay(), [query_opt()] ) :: [EctoPGMQ.Message.id()]
Sends the given messages to the given queue.
For more information about this function, see the PGMQ docs.
Examples
iex> [message_id] = PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> is_integer(message_id)
true
iex> delay = DateTime.utc_now()
iex> [message_id] = PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}], nil, delay)
iex> is_integer(message_id)
true@spec send_batch_topic( Ecto.Repo.t(), routing_key(), [payload() | nil], [EctoPGMQ.Message.headers() | nil] | nil, delay(), [query_opt()] ) :: queue_message_ids()
Sends the given messages with the given routing key.
For more information about message routing, see Message Routing.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.bind_topic(Repo, "#", "my_queue")
iex> %{"my_queue" => [message_id]} = PGMQ.send_batch_topic(Repo, "my.routing.key", [%{"id" => 1}])
iex> is_integer(message_id)
true
iex> delay = DateTime.utc_now()
iex> PGMQ.bind_topic(Repo, "#", "my_queue")
iex> %{"my_queue" => [message_id]} = PGMQ.send_batch_topic(Repo, "my.routing.key", [%{"id" => 1}], nil, delay)
iex> is_integer(message_id)
true@spec set_vt(Ecto.Repo.t(), EctoPGMQ.Queue.name(), [EctoPGMQ.Message.id()], delay(), [ query_opt() ]) :: [ EctoPGMQ.Message.t() ]
Sets the visibility timeout of the given messages in the given queue.
For more information about this function, see the PGMQ docs.
Examples
iex> message_ids = PGMQ.send_batch(Repo, "my_queue", [%{"id" => 1}])
iex> [read_message] = PGMQ.read(Repo, "my_queue", 5, 1)
iex> [updated_message] = PGMQ.set_vt(Repo, "my_queue", message_ids, 10)
iex> DateTime.diff(updated_message.visible_at, read_message.visible_at) > 0
true@spec test_routing(Ecto.Repo.t(), routing_key(), [query_opt()]) :: [ EctoPGMQ.Binding.t() ]
Returns all of the bindings that match the given routing key.
Because this function naively wraps the corresponding PGMQ function, the
:bound_at field in the returned EctoPGMQ.Binding structs will not be
populated.
For more information about message routing, see Message Routing.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.bind_topic(Repo, "#", "my_queue")
iex> [%Binding{pattern: "#", queue: "my_queue"}] = PGMQ.test_routing(Repo, "my.routing.key")@spec unbind_topic(Ecto.Repo.t(), EctoPGMQ.Binding.pattern(), EctoPGMQ.Queue.name(), [ query_opt() ]) :: :ok
Unbinds the given queue from the given pattern.
For more information about message routing, see Message Routing.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.bind_topic(Repo, "#", "my_queue")
iex> PGMQ.unbind_topic(Repo, "#", "my_queue")
:ok@spec update_notify_insert( Ecto.Repo.t(), EctoPGMQ.Queue.name(), throttle_interval(), [query_opt()] ) :: :ok
Updates the insert notification throttle for the given queue.
For more information about notifications, see EctoPGMQ.Notifications.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.enable_notify_insert(Repo, "my_queue")
iex> PGMQ.update_notify_insert(Repo, "my_queue", 500)
:ok@spec validate_routing_key(Ecto.Repo.t(), routing_key(), [query_opt()]) :: :ok
Validates the given routing key.
Invalid Routing Key
This function will raise a Postgrex.Error if an invalid routing key is
given.
For more information about message routing, see Message Routing.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.validate_routing_key(Repo, "my.routing.key")
:ok@spec validate_topic_pattern(Ecto.Repo.t(), EctoPGMQ.Binding.pattern(), [query_opt()]) :: :ok
Validates the given binding pattern.
Invalid Pattern
This function will raise a Postgrex.Error if an invalid pattern is given.
For more information about message routing, see Message Routing.
For more information about this function, see the PGMQ docs.
Examples
iex> PGMQ.validate_topic_pattern(Repo, "#")
:ok