# Queuetopia

[![GitHub Workflow Status](]( [![GitHub issues](]( [![License](]( [![](]( [![](](

A persistant blocking job queue built with Ecto.

#### Features

- Persistence — Jobs are stored in a DB and updated after each execution attempt.

- Blocking — A failing job blocks its queue until it is done.

- Dynamicity — Queues are dynamically defined. Once the first job is created
  for the queue, the queue exists.

- Reactivity — Immediatly try to execute a job that has just been created.

- Scheduled Jobs — Allow to schedule job in the future.

- Retries — Failed jobs are retried with a configurable backoff.

- Persistence — Jobs are stored in a DB and updated after each execution attempt.

- Performance — At each poll, only one job per queue is run. Optionnaly, jobs can
  avoid waiting unnecessarily. The performed job triggers an other polling.

- Isolated Queues — Jobs are stored in a single table but are executed in
  distinct queues. Each queue runs in isolation, ensuring that a job in a single
  slow queue can't back up other faster queues and that a failing job in a queue
  don't block other queues.

- Handle Node Duplication — Queues are locked, preventing two nodes to perform
  the same job at the same time.

## Installation

Queuetopia is published on [Hex](
The package can be installed by adding `queuetopia` to your list of dependencies in `mix.exs`:

def deps do
    {:queuetopia, "~> 2.0.1"}

After the packages are installed you must create a database migration to
add the queuetopia tables to your database:

mix ecto.gen.migration create_queuetopia_tables

Open the generated migration in your editor and call the `up` and `down`
functions on `Queuetopia.Migrations`:

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

  def up do

  def down do

Now, run the migration to create the table:

mix ecto.migrate

## Usage

### Defining the Queuetopia

A Queuetopia must be informed a repo to persist the jobs and a performer module,
responsible to execute the jobs.

Define a Queuetopia with a repo and a perfomer like this:

defmodule MyApp.MailQueuetopia do
  use Queuetopia,
    repo: MyApp.Repo,
    performer: MyApp.MailQueuetopia.Performer

Define the perfomer, adopting the Queuetopia.Performer behaviour, like this:

defmodule MyApp.MailQueuetopia.Performer do
  @behaviour Queuetopia.Performer

  @impl true
  def perform(%Queuetopia.Queue.Job{action: "do_x"}) do

  defp do_x(), do: {:ok, "done"}

### Start the Queuetopia

An instance Queuetopia is a supervision tree and can be started as a child of a supervisor.

For instance, in the application supervision tree:

defmodule MyApp do
  use Application

  def start(_type, _args) do
    children = [
    Supervisor.start_link(children, strategy: :one_for_one)

Or, it can be started directly like this:


The configuration can be set as below:

 # config/config.exs
  config :my_app, MyApp.MailQueuetopia,
    poll_interval: 60 * 1_000,
    disable?: true


Note that the polling interval is optionnal and is an available param of start_link/1.
By default, it will be set to 60 seconds.

### Feeds your queues

To create a job defines its action and its params and configure its timeout and the max backoff for the retries.
By default, the job timeout is set to 60 seconds, the max backoff to 24 hours and the max attempts to 20.

MyApp.MailQueuetopia.create_job!("mails_queue_1", "send_mail", %{email_address: "", body: "Welcome"}, [timeout: 1_000, max_backoff: 60_000])

MyApp.MailQueuetopia.create_job("mails_queue_1", "send_mail", %{email_address: "", body: "Welcome"}, [timeout: 1_000, max_backoff: 60_000])
to handle changeset errors.

So, the mails_queue_1 was born and you can add it other jobs as we do above.
When the job creation is out of transaction, Queuetopia is automatically notified about the new job.
Anyway, you can notify the queuetopia about a new created job.


### One DB, many Queuetopia

Multiple Queuetopia can coexist in your project, e.g your project may own its Queuetopia and uses a library
shipping its Queuetopia. The both Queuetopia may run on the same DB and share the same repo. They will have a different scheduler,
may have a different polling interval. They will be defined a scope to reach only their own jobs,
so the won't interfer each other.

## Test

Rename env/test.env.example to env/test.env, set your params and source it.

MIX_ENV=test mix do ecto.drop, ecto.create, ecto.migrate
mix test

Documentation can be generated with [ExDoc](
and published on [HexDocs]( Once published, the docs can
be found at [](

Thanks to [Oban] [] and elixir community who inspired the Queuetopia development.