defmodule Queuetopia do @moduledoc """ Defines a queues machine. A Queuetopia can manage multiple ordered blocking queues. All the queues share only the same scheduler and the poll interval. They are completely independant from each other. A Queuetopia expects a performer to exist. For example, the performer can be implemented like this: defmodule MyApp.MailQueue.Performer do @behaviour Queuetopia.Jobs.Performer @impl true def perform(%Queuetopia.Jobs.Job{action: "do_x"}) do do_x() end defp do_x(), do: {:ok, "done"} end And the Queuetopia: defmodule MyApp.MailQueue do use Queuetopia, repo: MyApp.Repo, performer: MyApp.MailQueue.Performer end """ @doc """ Creates a job. ## Job options A job accepts the following options: * `:timeout` - The time in milliseconds to wait for the job to finish. (default: 60_000) * `:max_backoff` - default to 24 * 3600 * 1_000 * `:max_attempts` - default to 20. ## Examples iex> MyApp.MailQueue.create_job("mails_queue_1", "send_mail", %{email_address: "toto@mail.com", body: "Welcome"}, [timeout: 1_000, max_backoff: 60_000]) """ @callback create_job(binary(), binary(), map(), [Job.option()]) :: {:error, Ecto.Changeset.t()} | {:ok, Job.t()} defmacro __using__(opts) do quote bind_quoted: [opts: opts] do @behaviour Queuetopia use Supervisor alias Queuetopia.Jobs.Job @typedoc "Option values used by the `start*` functions" @type option :: {:poll_interval, non_neg_integer()} @repo Keyword.fetch!(opts, :repo) @performer Keyword.fetch!(opts, :performer) |> to_string() @scope __MODULE__ |> to_string() @default_poll_interval 60 * 1_000 @doc """ Starts the Queuetopia supervisor process. The :poll_interval can also be given in order to config the polling interval of the scheduler. """ @spec start_link([option()]) :: Supervisor.on_start() def start_link(opts \\ []) do poll_interval = Keyword.get(opts, :poll_interval, @default_poll_interval) repoll_after_job_performed? = Keyword.get(opts, :repoll_after_job_performed?, false) Supervisor.start_link( __MODULE__, [ repo: @repo, poll_interval: poll_interval, repoll_after_job_performed?: repoll_after_job_performed? ], name: __MODULE__ ) end @impl true def init(args) do children = [ {Task.Supervisor, name: child_name("TaskSupervisor")}, {Queuetopia.Scheduler, [ name: child_name("Scheduler"), task_supervisor_name: child_name("TaskSupervisor"), repo: Keyword.fetch!(args, :repo), repoll_after_job_performed?: Keyword.fetch!(args, :repoll_after_job_performed?), scope: @scope, poll_interval: Keyword.fetch!(args, :poll_interval) ]} ] Supervisor.init(children, strategy: :one_for_one) end defp child_name(child) do Module.concat(__MODULE__, child) end @spec create_job(binary(), binary(), map(), [Job.option()]) :: {:error, Ecto.Changeset.t()} | {:ok, Job.t()} def create_job(queue, action, params, opts \\ []) do Queuetopia.Jobs.create_job(@repo, @performer, @scope, queue, action, params, opts) end def repo() do @repo end def performer() do @performer end def scope() do @scope end end end end