defmodule Verk do @moduledoc """ Verk is a job processing system that integrates well with Sidekiq jobs Each queue will have a pool of workers handled by `poolboy` that will process jobs. Verk has a retry mechanism similar to Sidekiq that keeps retrying the jobs with a reasonable backoff. It has an API that provides information about the queues """ alias Verk.{Job, Time, Manager} @schedule_key "schedule" @doc """ Add a new `queue` with a pool of size `size` of workers """ @spec add_queue(atom, pos_integer) :: Supervisor.on_start_child def add_queue(queue, size \\ 25) when is_atom(queue) and size > 0 do Manager.add(queue, size) end @doc """ Remove `queue` from the list of queues that are being processed """ @spec remove_queue(atom) :: :ok | {:error, :not_found} def remove_queue(queue) when is_atom(queue) do Manager.remove(queue) end defdelegate pause_queue(queue), to: Verk.Manager, as: :pause defdelegate resume_queue(queue), to: Verk.Manager, as: :resume @doc """ Enqueues a Job to the specified queue returning the respective job id The job must have: * a valid `queue` * a list of `args` to perform * a module to perform (`class`) * a valid `jid` Optionally a Redix server can be passed which defaults to `Verk.Redis` """ @spec enqueue(%Job{}, GenServer.server) :: {:ok, binary} | {:error, term} def enqueue(job, redis \\ Verk.Redis) def enqueue(job = %Job{queue: nil}, _redis), do: {:error, {:missing_queue, job}} def enqueue(job = %Job{class: nil}, _redis), do: {:error, {:missing_module, job}} def enqueue(job = %Job{args: args}, _redis) when not is_list(args), do: {:error, {:missing_args, job}} def enqueue(job = %Job{max_retry_count: nil}, redis) do job = %Job{job | max_retry_count: Job.default_max_retry_count()} enqueue(job, redis) end def enqueue(job = %Job{max_retry_count: count}, _redis) when not is_integer(count), do: {:error, {:invalid_max_retry_count, job}} def enqueue(job = %Job{jid: nil}, redis), do: enqueue(%Job{job | jid: generate_jid()}, redis) def enqueue(job = %Job{jid: jid, queue: queue}, redis) do job = %Job{job | enqueued_at: Time.now |> DateTime.to_unix} case Redix.command(redis, ["LPUSH", "queue:#{queue}", Job.encode!(job)]) do {:ok, _} -> {:ok, jid} {:error, reason} -> {:error, reason} end end @doc """ Schedules a Job to the specified queue returning the respective job id The job must have: * a valid `queue` * a list of `args` to perform * a module to perform (`class`) * a valid `jid` Optionally a Redix server can be passed which defaults to `Verk.Redis` """ @spec schedule(%Job{}, %DateTime{}, GenServer.server) :: {:ok, binary} | {:error, term} def schedule(job, datetime, redis \\ Verk.Redis) def schedule(job = %Job{queue: nil}, %DateTime{}, _redis), do: {:error, {:missing_queue, job}} def schedule(job = %Job{class: nil}, %DateTime{}, _redis), do: {:error, {:missing_module, job}} def schedule(job = %Job{args: args}, %DateTime{}, _redis) when not is_list(args), do: {:error, {:missing_args, job}} def schedule(job = %Job{jid: nil}, perform_at = %DateTime{}, redis) do schedule(%Job{job | jid: generate_jid()}, perform_at, redis) end def schedule(job = %Job{jid: jid}, perform_at = %DateTime{}, redis) do if Time.after?(Time.now, perform_at) do #past time to do the job enqueue(job, redis) else case Redix.command(redis, ["ZADD", @schedule_key, DateTime.to_unix(perform_at), Job.encode!(job)]) do {:ok, _} -> {:ok, jid} {:error, reason} -> {:error, reason} end end end defp generate_jid do <> = :crypto.strong_rand_bytes(8) "#{part1}#{part2}" end end