EctoPGMQ.Message (ecto_pgmq v1.0.0)

Copy Markdown View Source

Schema for PGMQ messages.

Summary

Types

group()

A PGMQ message group.

headers()

PGMQ message headers.

id()

A PGMQ message ID.

payload()

A PGMQ message payload.

specification()

A PGMQ message specification.

t()

A PGMQ message.

Functions

archive_query(queue)

Returns a query for the messages in the archive for the given queue.

queue_query(queue, opts \\ [])

Returns a query for the messages in the given queue.

Types

group()

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

A PGMQ message group.

For more information about FIFO message groups, see FIFO Message Groups.

headers()

@type headers() :: %{optional(String.Chars.t()) => term()}

PGMQ message headers.

id()

@type id() :: pos_integer()

A PGMQ message ID.

payload()

@type payload() :: %{optional(String.Chars.t()) => term()}

A PGMQ message payload.

specification()

@type specification() ::
  payload()
  | {payload(), group() | nil}
  | {payload(), group() | nil, headers() | nil}

A PGMQ message specification.

Warning

If the group is not nil, it will override any group that may already be specified in the headers.

t()

@type t() :: %EctoPGMQ.Message{
  archived_at: DateTime.t() | nil,
  enqueued_at: DateTime.t(),
  group: group() | nil,
  headers: headers() | nil,
  id: id(),
  payload: payload(),
  reads: non_neg_integer(),
  visible_at: DateTime.t()
}

A PGMQ message.

Functions

archive_query(queue)

@spec archive_query(EctoPGMQ.Queue.name()) :: Ecto.Query.t()

Returns a query for the messages in the archive for the given queue.

Examples

iex> message_ids = EctoPGMQ.send_messages(Repo, "my_queue", [%{"foo" => 1}, %{"bar" => 2}])
iex> EctoPGMQ.archive_messages(Repo, "my_queue", message_ids)
iex> messages = Repo.all(archive_query("my_queue"))
iex> Enum.all?(messages, &is_struct(&1, EctoPGMQ.Message))
true

queue_query(queue, opts \\ [])

@spec queue_query(
  EctoPGMQ.Queue.name(),
  keyword()
) :: Ecto.Query.t()

Returns a query for the messages in the given queue.

Options

  • :archived_at? - An optional boolean/0 denoting whether or not to select a NULL :archived_at column. This can be used to make the query structure match that of archive_query/1. Defaults to false.

Examples

iex> EctoPGMQ.send_messages(Repo, "my_queue", [%{"foo" => 1}, %{"bar" => 2}])
iex> messages = Repo.all(queue_query("my_queue"))
iex> Enum.all?(messages, &is_struct(&1, EctoPGMQ.Message))
true