defmodule Poolder.Scheduler do defmacro __using__(opts) do name = Keyword.get(opts, :name, __MODULE__) jobs = Keyword.get(opts, :jobs, []) |> Enum.uniq_by(&elem(&1, 0)) retry = Keyword.get(opts, :retry, count: 0, backoff: 1000) retries = Keyword.get(retry, :count) backoff = Keyword.get(retry, :backoff, 0) hibernate_after = Keyword.get(opts, :hibernate_after, :infinity) priority = Keyword.get(opts, :priority, 0) quote bind_quoted: [ name: name, jobs: jobs, retries: retries, backoff: backoff, hibernate_after: hibernate_after, priority: priority ] do @name name @jobs jobs @retries retries @backoff backoff @catcher retries > 0 @hibernate_after hibernate_after @priority priority @priority_abnormal priority != :normal if @jobs == [] do raise "Scheduler #{name} has no jobs" end @behaviour Poolder.Scheduler def child_spec(args) do %{ id: @name, start: {__MODULE__, :start_link, [args]}, type: :worker, restart: :transient, shutdown: 500 } end def start(opts \\ []) do pid = spawn(__MODULE__, :run, [opts]) Process.register(pid, @name) {:ok, pid} end def start_link(opts \\ []) do pid = spawn_link(__MODULE__, :run, [opts]) Process.register(pid, @name) {:ok, pid} end def run(opts) do trefs = for {key, interval} <- @jobs, into: %{} do { key, send_after(key, interval) } end args = %{jobs: Map.new(@jobs), trefs: trefs, opts: opts} loop(args) end def loop(state, hibernate_after \\ @hibernate_after) do receive do {:kill, key} -> handle_info({:kill, key}, state) {:set, key, interval} -> handle_info({:set, key, interval}, state) {:timeout, key} -> handle_info({:timeout, key}, state) {:retry_job, key, attempt} -> handle_info({:retry_job, key, attempt}, state) _ -> :ignore after hibernate_after -> handle_hibernate(state) :erlang.hibernate(__MODULE__, :loop, [state]) end end defp handle_info({:kill, key}, state) do tref = Map.get(state.trefs, key) cancel_timer(tref) loop(%{state | trefs: Map.delete(state.trefs, key)}) end defp handle_info({:set, key, interval}, state) do tref = Map.get(state.trefs, key) cancel_timer(tref) tref = send_after(key, interval) loop(%{state | trefs: Map.put(state.trefs, key, tref)}) end defp handle_info({:timeout, key}, state = %{jobs: jobs, trefs: trefs}) do pid = self() interval = Map.get(jobs, key) tref = send_after(key, interval) spawn_link(__MODULE__, :try_run, [pid, key, 1, state]) loop(%{state | trefs: Map.put(trefs, key, tref)}) end defp handle_info({:retry_job, key, attempt}, state) do try_run(self(), key, attempt, state) loop(state) end ## Public API def stop(task) do send(__MODULE__, {:kill, task}) end def schedule(task, interval) do send(__MODULE__, {:set, task, interval}) end def try_run(pid, key, attempt, state) when attempt <= @retries do try do if @priority_abnormal, do: :erlang.process_flag(:priority, @priority) case :erlang.apply(__MODULE__, key, [state]) do {:set, new_interval} -> send(pid, {:set, key, new_interval}) {:set, name, new_interval} -> send(pid, {:set, name, new_interval}) {:kill, key} -> send(pid, {:kill, key}) :kill -> send(pid, {:kill, key}) :stop -> Process.exit(pid, :normal) _ -> :ok end rescue error -> @catcher and case handle_error(key, attempt, error, state) do {:retry, new_state} -> try_run(pid, key, attempt + 1, new_state) {:backoff, delay} -> Process.send_after(pid, {:retry_job, key, attempt + 1}, delay) _halt -> :ok end end end def try_run(_pid, _key, _attempt, _state), do: :ok if @backoff > 0 do def handle_error(_key, _attempt, _error, state), do: {:backoff, @backoff} else def handle_error(_key, _attempt, _error, state), do: {:retry, state} end def handle_hibernate(_state), do: :ok defoverridable stop: 1, schedule: 2, handle_error: 4, handle_hibernate: 1 ## Private API defp cancel_timer(nil), do: :ok defp cancel_timer(tref), do: Process.cancel_timer(tref) defp send_after(key, interval) do case next_interval(interval) do :error -> :ignore next_interval -> Process.send_after(self(), {:timeout, key}, next_interval) end end defp next_interval(cron) when is_binary(cron) do case Cron.new(cron) do {:ok, cron} -> Cron.until(cron, DateTime.utc_now()) {:error, _} -> :error end end defp next_interval(x) when is_integer(x), do: x defp next_interval(:year), do: 31_536_000_000 defp next_interval(:month), do: 2_592_000_000 defp next_interval(:week), do: 604_800_000 defp next_interval(:day), do: 86_400_000 defp next_interval(:hour), do: 3_600_000 defp next_interval(:minute), do: 60_000 defp next_interval(:second), do: 1_000 defp next_interval(_), do: :error end end @callback handle_hibernate(state :: any) :: any @callback handle_error(job :: term, attempt :: integer, error :: any, state :: any) :: {:retry, new_state :: any} | {:backoff, delay :: integer} | :halt end