GenQueue Oban

This is an adapter for GenQueue to enable functionality with Oban.

Installation

The package can be installed by adding gen_queue_oban to your list of dependencies in mix.exs:

defdepsdo[{:gen_queue_oban,"~> 0.1.0"}]end

Documentation

See HexDocs for additional documentation.

Configuration

Before starting, please refer to the Oban documentation for details on configuration. This adapter handles zero Oban related config.

Creating Enqueuers

We can start off by creating a new GenQueue module, which we will use to push jobs to Oban.

defmoduleEnqueuerdouseGenQueue,otp_app: :my_append

Once we have our module setup, ensure we have our config pointing to the GenQueue.Adapters.Oban adapter.

config:my_app,Enqueuer,[adapter: GenQueue.Adapters.Oban]

Starting Enqueuers

By default, gen_queue_oban does not start Oban on application start. So we must add our new Enqueuer module to our supervision tree.

children=[supervisor(Enqueuer,[]),]

Creating Jobs

Jobs are simply modules with a perform method. With Oban we must add use Oban.Worker to our jobs with the relevant configuration.

defmoduleMyJobdouseOban.Worker,queue: "events",max_attempts: 10@implOban.Workerdefperform(arg1,_job)doIO.inspect(arg1)endend

Enqueuing Jobs

We can now easily enqueue jobs to Oban. The adapter will handle a variety of argument formats.

# Please note that zero-arg jobs default to using %{}, as per Oban requirements.# Push MyJob to your default queue with %{} arg.{:ok,job}=Enqueuer.push(MyJob)# Push MyJob to your default queue with %{} arg.{:ok,job}=Enqueuer.push({MyJob})# Push MyJob to your default queue with %{"foo" => "bar"} arg.{:ok,job}=Enqueuer.push({MyJob,%{"foo"=>"bar"}})# Push MyJob to "default" queue with %{} arg.{:ok,job}=Enqueuer.push({MyJob,[]})# Push MyJob to "default" queue with %{"foo" => "bar"} arg.{:ok,job}=Enqueuer.push({MyJob,[%{"foo"=>"bar"}]})# Push MyJob to "foo" queue with %{"foo" => "bar"} arg{:ok,job}=Enqueuer.push({MyJob,%{"foo"=>"bar"}},[queue: "foo"])# Schedule MyJob to your default queue with %{"foo" => "bar"} arg in 10 seconds{:ok,job}=Enqueuer.push({MyJob,%{"foo"=>"bar"}},[delay: 10_000])# Schedule MyJob to your default queue with %{"foo" => "bar"} arg at a specific timedate=DateTime.utc_now(){:ok,job}=Enqueuer.push({MyJob,%{"foo"=>"bar"}},[delay: date])

Testing

Optionally, we can also have our tests use the GenQueue.Adapters.MockJob adapter.

config:my_app,Enqueuer,[adapter: GenQueue.Adapters.MockJob]

This mock adapter uses the standard GenQueue.Test helpers to send the job payload back to the current processes mailbox (or another named process) instead of actually enqueuing the job to rabbitmq.

defmoduleMyJobTestdouseExUnit.Case,async: trueimportGenQueue.Testsetupdosetup_test_queue(Enqueuer)endtest"my enqueuer works"do{:ok,_}=Enqueuer.push(Job)assert_receive(%GenQueue.Job{module: Job,args: []})endend

If your jobs are being enqueued outside of the current process, we can use named processes to recieve the job. This wont be async safe.

importGenQueue.Testsetupdosetup_global_test_queue(Enqueuer,:my_process_name)end