Graham McIntire
0db5c2ae3e
Adds the commercial Oban Pro package (vendored from the towerops-web2 vendor tree) so this project can use its workers, plugins, and Smart engine features.
111 lines
2.7 KiB
Elixir
111 lines
2.7 KiB
Elixir
defmodule Oban.Pro.Refresher do
|
|
@moduledoc false
|
|
|
|
# Centralized refreshing and cleaning of producer records across all Oban instances.
|
|
#
|
|
# The Refresher runs as a single process supervised by the Oban.Pro application. It refreshes
|
|
# producer records for all actively running queues on the current node.
|
|
|
|
use GenServer
|
|
|
|
import Ecto.Query, only: [where: 3]
|
|
|
|
alias __MODULE__, as: State
|
|
alias Oban.Pro.Producer
|
|
alias Oban.{Peer, Repo}
|
|
|
|
defstruct [
|
|
:timer,
|
|
interval: :timer.seconds(15),
|
|
producers: %{},
|
|
producer_ttl: :timer.minutes(1)
|
|
]
|
|
|
|
@doc false
|
|
def start_link(opts \\ []) do
|
|
state = struct!(State, opts)
|
|
|
|
GenServer.start_link(__MODULE__, state, name: __MODULE__)
|
|
end
|
|
|
|
# GenServer Callbacks
|
|
|
|
@impl GenServer
|
|
def init(state) do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
{:ok, schedule_refresh(state)}
|
|
end
|
|
|
|
@impl GenServer
|
|
def terminate(_reason, state) do
|
|
if is_reference(state.timer), do: Process.cancel_timer(state.timer)
|
|
|
|
:ok
|
|
end
|
|
|
|
@impl GenServer
|
|
def handle_call(:refresh, _from, state) do
|
|
{:noreply, state} = handle_info(:refresh, state)
|
|
|
|
{:reply, :ok, state}
|
|
end
|
|
|
|
@impl GenServer
|
|
def handle_info(:refresh, state) do
|
|
refresh_producers(state)
|
|
cleanup_producers(state)
|
|
|
|
{:noreply, schedule_refresh(state)}
|
|
end
|
|
|
|
# Helpers
|
|
|
|
defp schedule_refresh(state) do
|
|
timer = Process.send_after(self(), :refresh, state.interval)
|
|
|
|
%{state | timer: timer}
|
|
end
|
|
|
|
defp refresh_producers(_state) do
|
|
match = [{{{:"$1", {:producer, :"$2"}}, :_, :_}, [], [[:"$1", :"$2"]]}]
|
|
|
|
Oban.Registry
|
|
|> Registry.select(match)
|
|
|> Enum.group_by(&hd/1, &Enum.at(&1, 1))
|
|
|> Enum.each(fn {oban_name, queues} ->
|
|
with {_pid, %{testing: :disabled} = conf} <- Oban.Registry.lookup(oban_name) do
|
|
now = DateTime.utc_now()
|
|
ids = Enum.flat_map(queues, &lookup_uuid(&1, oban_name))
|
|
|
|
refresh_query = where(Producer, [p], p.uuid in ^ids)
|
|
|
|
Repo.update_all(conf, refresh_query, set: [updated_at: now])
|
|
end
|
|
end)
|
|
end
|
|
|
|
defp cleanup_producers(state) do
|
|
match = [{{:"$1", :_, :"$2"}, [{:not, {:is_tuple, :"$1"}}], [:"$2"]}]
|
|
|
|
Oban.Registry
|
|
|> Registry.select(match)
|
|
|> Enum.filter(&Peer.leader?/1)
|
|
|> Enum.each(fn conf ->
|
|
now = DateTime.utc_now()
|
|
outdated_at = DateTime.add(now, -state.producer_ttl, :millisecond)
|
|
cleanup_query = where(Producer, [p], p.updated_at <= ^outdated_at)
|
|
|
|
Repo.delete_all(conf, cleanup_query)
|
|
end)
|
|
end
|
|
|
|
defp lookup_uuid(queue, oban_name) do
|
|
prod_name = {oban_name, {:producer, queue}}
|
|
|
|
case Registry.meta(Oban.Registry, prod_name) do
|
|
{:ok, %{uuid: uuid}} -> [uuid]
|
|
_error -> []
|
|
end
|
|
end
|
|
end
|