Defines a Worker for processing Jobs.

The defined worker is responsible for processing passed jobs, and handling the job's success and failure callbacks. The defined worker must export a perform/1 callback otherwise compilation will fail.

Basic Worker

defmodule MyApp.Workers.SignupMailer do
  use Que.Worker

  def perform(email) do
    Mailer.send_email(to: email, message: "Thank you for signing up!")
  end
end

You can also pattern match and use guard clauses like normal methods:

defmodule MyApp.Workers.NotificationSender do
  use Que.Worker

  def perform(type: :like, to: user, count: count) do
    User.notify(user, "You have #{count} new likes on your posts")
  end

  def perform(type: :message, to: user, from: sender) do
    User.notify(user, "You received a new message from #{sender.name}")
  end

  def perform(to: user) do
    User.notify(user, "New activity on your profile")
  end
end

Concurrency

By default, workers process one Job at a time. You can specify a custom value by passing the concurrency option.

defmodule MyApp.Workers.PageScraper do
  use Que.Worker, concurrency: 4

  def perform(url), do: Scraper.scrape(url)
end

If you want all Jobs to be processed concurrently without any limit, you can set the concurrency option to :infinity. The concurrency option must either be a positive integer or :infinity, otherwise it will raise an error during compilation.

Handle Job Success & Failure

The worker can also export optional on_success/1 and on_failure/2 callbacks that handle appropriate cases.

defmodule MyApp.Workers.CampaignMailer do
  use Que.Worker

  def perform({campaign, user}) do
    Mailer.send_campaign_email(campaign, user: user)
  end

  def on_success({campaign, user}) do
    CampaignReport.compile(campaign, status: :success, user: user)
  end

  def on_failure({campaign, user}, error) do
    CampaignReport.compile(campaign, status: :failed, user: user)
    Logger.debug("Campaign email to #{user.id} failed: #{inspect(error)}")
  end
end

Setup and Teardown

You can similarly export optional on_setup/1 and on_teardown/1 callbacks that are respectively run before and after the job is performed (successfully or not). But instead of the job arguments, they pass the job struct as an argument which holds a lot more internal details that can be useful for custom features such as logging, metrics, requeuing and more.

defmodule MyApp.Workers.VideoProcessor do
  use Que.Worker

  def on_setup(%Que.Job{} = job) do
    VideoMetrics.record(job.id, :start, process: job.pid, status: :starting)
  end

  def perform({user, video, options}) do
    User.notify(user, "Your video is processing, check back later.")
    FFMPEG.process(video.path, options)
  end

  def on_teardown(%Que.Job{} = job) do
    {user, video, _options} = job.arguments
    link = MyApp.Router.video_path(user.id, video.id)

    VideoMetrics.record(job.id, :end, status: job.status)
    User.notify(user, "We've finished processing your video. See the results.", link)
  end
end

Failed Job Retries

Failed Jobs are NOT automatically retried. If you want a job to be retried when it fails, you can simply enqueue it again.

To get a list of all failed jobs, you can call Que.Persistence.failed/0.