Packages
oban
2.13.2
2.23.0
2.22.1
2.22.0
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.4
2.19.3
2.19.2
2.19.1
2.19.0
2.18.3
2.18.2
2.18.1
2.18.0
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.3
2.16.2
2.16.1
2.16.0
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.2
2.14.1
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.3
2.11.2
2.11.1
2.11.0
2.10.1
2.10.0
retired
2.9.2
2.9.1
2.9.0
2.8.0
2.7.2
2.7.1
2.7.0
2.6.1
2.6.0
2.5.0
2.4.3
2.4.2
2.4.1
2.4.0
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
1.2.0
1.1.0
1.0.0
1.0.0-rc.2
1.0.0-rc.1
0.12.1
0.12.0
0.11.1
0.11.0
0.10.1
0.10.0
0.9.0
0.8.1
0.8.0
0.7.1
0.7.0
0.6.0
0.5.0
0.4.0
0.3.0
0.2.0
0.1.0
Robust job processing, backed by modern PostgreSQL, SQLite3, and MySQL.
Current section
Files
Jump to
Current section
Files
lib/oban/queue/watchman.ex
defmodule Oban.Queue.Watchman do
@moduledoc false
use GenServer
alias Oban.Queue.Producer
@type option ::
{:name, module()}
| {:foreman, identifier()}
| {:producer, identifier()}
| {:shutdown, timeout()}
defmodule State do
@moduledoc false
defstruct [:foreman, :producer, :shutdown]
end
@spec child_spec([option]) :: Supervisor.child_spec()
def child_spec(opts) do
shutdown =
case opts[:shutdown] do
0 ->
:brutal_kill
value ->
value
end
%{id: __MODULE__, start: {__MODULE__, :start_link, [opts]}, shutdown: shutdown}
end
@spec start_link([option]) :: GenServer.on_start()
def start_link(opts) do
{name, opts} = Keyword.pop(opts, :name)
GenServer.start_link(__MODULE__, Map.new(opts), name: name)
end
@impl GenServer
def init(%{foreman: foreman, producer: producer}) do
Process.flag(:trap_exit, true)
{:ok, %State{foreman: foreman, producer: producer}}
end
@impl GenServer
def terminate(_reason, %State{foreman: foreman, producer: producer}) do
try do
:ok = Producer.shutdown(producer)
:ok = wait_for_executing(foreman)
catch
:exit, _reason -> :ok
end
:ok
end
defp wait_for_executing(foreman, interval \\ 50) do
# There is a chance that the consumer process doesn't exist, and we never want to raise
# another error as part of the shut down process.
children =
try do
Supervisor.count_children(foreman)
catch
_ -> %{active: 0}
end
case children do
%{active: 0} ->
:ok
_ ->
:ok = Process.sleep(interval)
wait_for_executing(foreman, interval)
end
end
end