EctoPGMQ

An opinionated PGMQ client for Elixir.

Requirements

The following requirements must be met to use this package in your project:

• This package requires an existing Ecto.Repo in order to execute PostgreSQL queries. The aforementioned Ecto.Repo must use the Ecto.Adapters.Postgres adapter.

• While EctoPGMQ.Migrations contains helper functions for managing the PGMQ extension, this package does NOT include the PGMQ extension. For additional information about installing PGMQ, see PGMQ Installation.

Installation

This package requires that your project is already using Ecto with the Ecto.Adapters.Postgres adapter. If that's the case, this package can be installed by adding :ecto_pgmq to your list of dependencies in mix.exs:

defp deps do
  [
    # `Ecto` and `Postgrex` should already be present
    {:ecto_sql, "~> 3.11"},
    {:postgrex, ">= 0.0.0"},
    # Broadway is required to use the `EctoPGMQ.Producer` module
    {:broadway, "~> 1.0"},
    {:ecto_pgmq, "~> 2.0"}
  ]
end

Documentation

For additional documentation, see HexDocs.

Usage

EctoPGMQ is designed to be used seamlessly with your application repo. Some common operations are showcased below but the full feature-space can be found in the Documentation.

Installing the PGMQ Extension

The most common way to install the PGMQ extension is in a database migration. We can create a database migration with the following command:

mix ecto.gen.migration create_pgmq

We can then implement the migration accordingly:

defmodule MyApp.Repo.Migrations.CreatePgmq do
  use Ecto.Migration

  alias EctoPGMQ.Migrations

  def change, do: Migrations.create_extension()
end

For more information about installing PGMQ, see PGMQ Installation.

Creating a Queue

The most common way to create a queue is in a database migration. We can create a database migration with the following command:

mix ecto.gen.migration create_my_queue

We can then implement the migration accordingly:

defmodule MyApp.Repo.Migrations.CreateMyQueue do
  use Ecto.Migration

  alias EctoPGMQ.Migrations

  def change, do: Migrations.create_queue("my_queue")
end

Sending Messages

Messages can be sent to a queue with the following function:

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

Reading Messages

Messages can be read from a queue with the following function:

EctoPGMQ.read_messages(MyApp.Repo, "my_queue", 300, 2)

Alternatively, messages can also be consumed in a Broadway pipeline with the EctoPGMQ.Producer module.

Dropping a Queue

The most common way to drop a queue is in a database migration. We can create a database migration with the following command:

mix ecto.gen.migration drop_my_queue

We can then implement the migration accordingly:

defmodule MyApp.Repo.Migrations.DropMyQueue do
  use Ecto.Migration

  alias EctoPGMQ.Migrations

  def change, do: Migrations.drop_queue("my_queue")
end