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) quote bind_quoted: [name: name, jobs: jobs, retries: retries, backoff: backoff] do @name name @jobs jobs @retries retries @backoff backoff @catcher retries > 0 use GenServer @behaviour Poolder.Scheduler def child_spec(args) do %{ id: @name, start: {__MODULE__, :start_link, [args]}, type: :worker, restart: :transient, shutdown: 500 } end def start_link(args) do if @jobs != [] do GenServer.start_link(__MODULE__, args, name: @name) else :ignore end end @impl true def init(args) do trefs = for {key, interval} <- @jobs, into: %{} do { key, send_after(key, interval) } end {:ok, %{jobs: Map.new(@jobs), trefs: trefs, args: args}} end @impl true def handle_info({:kill, key}, state) do tref = Map.get(state.trefs, key) cancel_timer(tref) {:noreply, %{state | trefs: Map.delete(state.trefs, key)}} end def handle_info({:set, key, interval}, state) do tref = Map.get(state.trefs, key) cancel_timer(tref) tref = send_after(key, interval) {:noreply, %{state | trefs: Map.put(state.trefs, key, tref)}} end def 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]) {:noreply, %{state | trefs: Map.put(trefs, key, tref)}} end def handle_info({:retry_job, key, attempt}, state) do try_run(self(), key, attempt, 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) do try do 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 -> GenServer.stop(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) {:noreply, state} :halt -> {:noreply, state} _ -> {:noreply, state} end) || {:noreply, state} end end 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 defoverridable stop: 1, schedule: 2, handle_error: 4 ## 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_error(job :: term, attempt :: integer, error :: any, state :: any) :: {:retry, new_state :: any} | {:backoff, delay :: integer} | :halt end