oban

Robust job processing in Elixir, backed by modern PostgreSQL. Reliable,
observable and loaded with enterprise grade features.

Hex Version Hex Docs

CircleCI Status MIT License

Table of Contents

Features

Oban's primary goals are reliability, consistency and observability. It is fundamentally different from other background job processing tools because it retains job data for historic metrics and inspection. You can leave your application running indefinitely without worrying about jobs being lost or orphaned due to crashes.

Advantages over in-memory, mnesia, Redis and RabbitMQ based tools:

Advanced features and advantages over other RDBMS based tools:

Installation

Oban is published on Hex. Add it to your list of dependencies in mix.exs:

def deps do

  [

    {:oban, "~> 0.2"}

  ]

end

Then run mix deps.get to install Oban and its dependencies, including Ecto, Jason and Postgrex.

After the packages are installed you must create a database migration to add the oban_jobs table to your database:

mix ecto.gen.migration add_oban_jobs_table

Open the generated migration in your editor and delegate the up and down functions to Oban.Migrations:

defmodule MyApp.Repo.Migrations.AddObanJobsTable do

  use Ecto.Migration



  defdelegate up, to: Oban.Migrations

  defdelegate down, to: Oban.Migrations

end

This will run all of Oban's versioned migrations for your database. Migrations between versions are idempotent and will never change after a release. As new versions are released you may need to run additional migrations.

Now, run the migration to create the table:

mix ecto.migrate

Next see Usage for how to integrate Oban into your application and start defining jobs!

Usage

Oban isn't an application and won't be started automatically. It is started by a supervisor that must be included in your application's supervision tree. All of your configuration is passed into the Oban supervisor, allowing you to configure Oban like the rest of your application.

# confg/config.exs

config :my_app, Oban,

  repo: MyApp.Repo,

  prune: {:maxlen, 100_000},

  queues: [default: 10, events: 50, media: 20]



# lib/my_app/application.ex

defmodule MyApp.Application do

  @moduledoc false



  use Application



  alias MyApp.{Endpoint, Repo}



  def start(_type, _args) do

    children = [

      Repo,

      Endpoint,

      {Oban, Application.get_env(:my_app, Oban)}

    ]



    Supervisor.start_link(children, strategy: :one_for_one, name: MyApp.Supervisor)

  end

end

If you are running tests (which you should be) you'll need to disable pruning and job dispatching altogether when testing:

# config/test.exs

config :my_app, Oban,

  queues: false,

  prune: :disabled

Without dispatch and pruning disabled Ecto will raise constant ownership errors and you won't be able to run tests.

Configuring Queues

Queues are specified as a keyword list where the key is the name of the queue and the value is the maximum number of concurrent jobs. The following configuration would start four queues with concurrency ranging from 5 to 50:

queues: [default: 10, mailers: 20, events: 50, media: 5]

There isn't a limit to the number of queues or how many jobs may execute concurrently. Here are a few caveats and guidelines:

Creating Workers

Worker modules do the work of processing a job. At a minimum they must define a perform/1 function, which is called with an args map.

Define a worker to process jobs in the events queue:

defmodule MyApp.Workers.Business do

  use Oban.Worker, queue: "events", max_attempts: 10



  @impl Oban.Worker

  def perform(%{"id" => id}) do

    model = MyApp.Repo.get(MyApp.Business.Man, id)



    IO.inspect(model)

  end

end

The return value of perform/1 doesn't matter and is entirely ignored. If the job raises an exception or throws an exit then the error will be reported and the job will be retried (provided there are attempts remaining).

Enqueueing Jobs

Jobs are simply Ecto strucs and are enqueued by inserting them into the database. Here we insert a job into the default queue and specify the worker by module name:

%{id: 1, user_id: 2}

|> Oban.Job.new(queue: :default, worker: MyApp.Worker)

|> MyApp.Repo.insert()

For convenience and consistency all workers implement a new/2 function that converts an args map into a job changeset suitable for inserting into the database:

%{in_the: "business", of_doing: "business"}

|> MyApp.Workers.Business.new()

|> MyApp.Repo.insert()

The worker's defaults may be overridden by passing options:

%{vote_for: "none of the above"}

|> MyApp.Workers.Business.new(queue: "special", max_attempts: 5)

|> MyApp.Repo.insert()

Jobs may be scheduled down to the second any time in the future:

%{id: 1}

|> MyApp.Workers.Business.new(schedule_in: 5)

|> MyApp.Repo.insert()

Testing

As noted in the Usage section above there are some guidelines for running tests:

Oban provides some helpers to facilitate testing. The helpers handle the boilerplate of making assertions on which jobs are enqueued. To use the assert_enqueued/1 and refute_enqueued/1 helpers in your tests you must include them in your testing module and specify your app's Ecto repo:

use Oban.Testing, repo: MyApp.Repo

Now you can assert or refute that jobs have been enqueued within your tests:

assert_enqueued worker: MyWorker, args: %{id: 1}



# or



refute_enqueued queue: "special", args: %{id: 2}

See the Oban.Testing module for more details.

Contributing

To run the Oban test suite you must have PostgreSQL 10+ running locally with a database named oban_test. Follow these steps to create the database, create the database and run all migrations:

# Create the database

MIX_ENV=test mix ecto.create -r Oban.Test.Repo



# Run the base migration

MIX_ENV=test mix ecto.migrate -r Oban.Test.Repo

To ensure a commit passes CI you should run mix ci locally, which executes the following commands: