Packages
ash
1.33.0
3.29.3
3.29.2
3.29.1
3.29.0
3.28.0
3.27.8
3.27.7
3.27.6
3.27.5
3.27.4
3.27.3
3.27.2
3.27.1
3.27.0
3.26.0
3.25.2
3.25.1
3.25.0
retired
3.24.7
3.24.6
3.24.5
3.24.4
3.24.3
3.24.2
3.24.1
3.24.0
3.23.1
3.23.0
3.22.2
3.22.1
3.22.0
3.21.3
3.21.2
3.21.1
3.21.0
3.20.0
3.19.3
3.19.2
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.0
3.15.0
3.14.1
3.14.0
retired
3.13.2
3.13.1
3.13.0
3.12.0
3.11.3
3.11.2
3.11.1
3.11.0
3.10.1
3.10.0
3.9.0
3.8.0
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
retired
3.6.3
retired
3.6.2
3.6.1
3.6.0
3.5.43
3.5.42
3.5.41
3.5.40
3.5.39
3.5.38
3.5.37
3.5.36
3.5.35
3.5.34
3.5.33
3.5.32
3.5.31
3.5.30
3.5.29
3.5.28
3.5.27
3.5.26
3.5.25
3.5.24
3.5.23
3.5.22
3.5.21
3.5.20
3.5.19
3.5.18
3.5.17
3.5.16
3.5.15
3.5.14
3.5.13
3.5.12
3.5.11
3.5.10
3.5.9
3.5.8
3.5.7
3.5.6
3.5.5
3.5.4
3.5.3
3.5.2
3.5.1
3.5.0
3.4.74
retired
3.4.73
3.4.72
3.4.71
3.4.70
3.4.69
3.4.68
3.4.67
3.4.66
3.4.65
3.4.64
3.4.63
3.4.62
3.4.61
3.4.60
3.4.59
3.4.58
3.4.57
3.4.56
3.4.55
3.4.54
3.4.53
3.4.52
3.4.51
3.4.50
3.4.49
3.4.48
3.4.47
3.4.46
3.4.45
3.4.44
3.4.43
3.4.42
3.4.41
3.4.40
3.4.39
3.4.38
3.4.37
3.4.36
3.4.35
3.4.34
3.4.33
3.4.32
3.4.31
3.4.30
3.4.29
3.4.28
3.4.27
3.4.26
3.4.25
3.4.24
3.4.23
3.4.22
3.4.21
3.4.20
3.4.19
3.4.18
3.4.17
3.4.16
3.4.15
3.4.14
3.4.13
3.4.12
3.4.11
3.4.10
3.4.9
3.4.8
3.4.7
3.4.6
3.4.5
3.4.4
3.4.3
3.4.2
3.4.1
3.4.0
3.3.3
3.3.2
3.3.1
3.3.0
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.16
3.0.15
3.0.14
3.0.13
3.0.12
3.0.11
3.0.10
3.0.9
3.0.8
3.0.7
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
3.0.0-rc.46
3.0.0-rc.45
3.0.0-rc.44
3.0.0-rc.43
3.0.0-rc.42
3.0.0-rc.41
3.0.0-rc.40
3.0.0-rc.39
3.0.0-rc.38
3.0.0-rc.37
3.0.0-rc.36
3.0.0-rc.35
3.0.0-rc.34
3.0.0-rc.33
3.0.0-rc.32
3.0.0-rc.31
3.0.0-rc.29
3.0.0-rc.28
3.0.0-rc.27
3.0.0-rc.26
3.0.0-rc.25
3.0.0-rc.24
3.0.0-rc.23
3.0.0-rc.22
3.0.0-rc.21
3.0.0-rc.20
3.0.0-rc.19
3.0.0-rc.18
3.0.0-rc.17
3.0.0-rc.16
3.0.0-rc.15
3.0.0-rc.14
3.0.0-rc.13
3.0.0-rc.12
3.0.0-rc.11
3.0.0-rc.10
3.0.0-rc.9
3.0.0-rc.8
3.0.0-rc.7
3.0.0-rc.6
3.0.0-rc.5
3.0.0-rc.4
3.0.0-rc.3
3.0.0-rc.1
3.0.0-rc.0
2.21.15
2.21.14
2.21.13
2.21.12
2.21.11
2.21.10
2.21.9
2.21.8
2.21.7
2.21.6
2.21.5
2.21.4
2.21.3
2.21.2
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.14
2.19.13
2.19.12
2.19.11
2.19.10
2.19.9
2.19.8
2.19.7
2.19.6
2.19.5
2.19.4
2.19.3
retired
2.19.2
retired
2.19.1
retired
2.19.0
retired
2.18.2
2.18.1
2.18.0
2.17.24
2.17.23
2.17.22
2.17.21
2.17.20
2.17.19
2.17.18
2.17.17
2.17.16
2.17.15
2.17.14
2.17.13
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.1
2.16.0
2.15.20
2.15.19
2.15.18
2.15.17
2.15.16
2.15.15
2.15.14
2.15.13
2.15.12
2.15.11
2.15.10
2.15.9
2.15.8
2.15.7
2.15.6
2.15.5
2.15.4
2.15.2
2.15.1
2.15.0
2.14.21
2.14.20
2.14.19
2.14.18
2.14.17
2.14.16
2.14.15
2.14.14
2.14.13
2.14.12
2.14.11
2.14.10
2.14.9
2.14.8
2.14.7
2.14.6
2.14.5
2.14.4
2.14.3
2.14.2
2.14.1
2.14.0
2.13.4
retired
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.11
2.11.10
2.11.9
2.11.8
2.11.7
2.11.6
2.11.5
2.11.4
2.11.3
2.11.2
2.11.1
2.11.0
2.11.0-rc.3
2.11.0-rc.2
2.11.0-rc.1
2.11.0-rc.0
2.10.2
2.10.1
2.10.0
2.9.29
2.9.28
2.9.27
2.9.26
2.9.25
2.9.24
2.9.23
2.9.22
2.9.21
2.9.20
2.9.19
2.9.18
2.9.17
2.9.16
2.9.15
2.9.14
2.9.13
2.9.12
2.9.11
2.9.10
2.9.9
2.9.8
2.9.7
2.9.6
2.9.5
2.9.4
2.9.3
2.9.2
2.9.1
2.9.0
2.8.1
2.8.0
2.7.1
2.7.0
2.6.31
2.6.30
2.6.29
2.6.28
2.6.27
2.6.26
2.6.25
2.6.24
2.6.23
2.6.22
2.6.21
2.6.20
2.6.19
2.6.18
2.6.17
2.6.16
2.6.15
2.6.14
2.6.13
2.6.11
2.6.10
2.6.9
2.6.8
2.6.7
2.6.6
2.6.5
2.6.4
2.6.3
2.6.2
2.6.1
2.6.0
2.5.16
2.5.15
2.5.14
2.5.13
2.5.12
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.5.0-rc.6
2.5.0-rc.5
2.5.0-rc.4
2.5.0-rc.3
2.5.0-rc.2
2.5.0-rc.1
2.5.0-rc.0
2.4.30
2.4.29
2.4.28
2.4.27
2.4.26
2.4.25
2.4.24
2.4.23
2.4.22
2.4.21
2.4.20
2.4.19
2.4.18
2.4.17
2.4.16
2.4.15
2.4.14
2.4.13
2.4.12
2.4.11
2.4.10
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.15
2.0.0-rc.14
2.0.0-rc.13
2.0.0-rc.12
2.0.0-rc.11
2.0.0-rc.10
2.0.0-rc.9
2.0.0-rc.8
2.0.0-rc.7
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
2.0.0-pre.8
2.0.0-pre.7
2.0.0-pre.6
2.0.0-pre.5
2.0.0-pre.4
2.0.0-pre.3
2.0.0-pre.2
2.0.0-pre.1
2.0.0-pre.0
1.53.3
1.53.2
1.53.0
1.52.0-rc.22
1.52.0-rc.21
1.52.0-rc.20
1.52.0-rc.19
1.52.0-rc.18
1.52.0-rc.17
1.52.0-rc.16
1.52.0-rc.15
1.52.0-rc.14
1.52.0-rc.13
1.52.0-rc.12
1.52.0-rc.11
1.52.0-rc.10
1.52.0-rc.9
1.52.0-rc.8
1.52.0-rc.7
1.52.0-rc.6
1.52.0-rc.5
1.52.0-rc.4
1.52.0-rc.3
1.52.0-rc.2
1.52.0-rc.1
1.52.0-rc.0
1.51.2
1.51.1
retired
1.51.0
1.50.21
1.50.20
1.50.19
1.50.18
1.50.17
1.50.16
1.50.15
1.50.14
1.50.13
1.50.12
1.50.11
1.50.10
1.50.9
1.50.8
1.50.7
1.50.6
1.50.5
1.50.4
1.50.3
1.50.2
1.50.1
1.50.0
1.49.0
1.48.0-rc.30
1.48.0-rc.29
1.48.0-rc.28
1.48.0-rc.27
1.48.0-rc.26
1.48.0-rc.25
1.48.0-rc.24
1.48.0-rc.23
1.48.0-rc.22
1.48.0-rc.21
1.48.0-rc.20
1.48.0-rc.19
1.48.0-rc.18
1.48.0-rc.17
1.48.0-rc.16
1.48.0-rc.15
1.48.0-rc.14
1.48.0-rc.13
1.48.0-rc.12
1.48.0-rc.11
1.48.0-rc.10
1.48.0-rc.9
1.48.0-rc.8
1.48.0-rc.7
1.48.0-rc.6
1.48.0-rc.5
1.48.0-rc.4
1.48.0-rc.3
1.48.0-rc.2
1.48.0-rc.1
1.48.0-rc.0
1.47.12
1.47.11
1.47.10
1.47.9
1.47.8
1.47.7
1.47.6
1.47.5
1.47.4
1.47.3
1.47.2
1.47.1
1.47.0
1.46.13
1.46.12
1.46.11
1.46.10
1.46.9
1.46.8
1.46.7
1.46.6
1.46.5
1.46.4
1.46.3
1.46.2
1.46.1
1.46.0
1.45.0-rc9
1.45.0-rc8
1.45.0-rc7
1.45.0-rc6
1.45.0-rc5
1.45.0-rc4
1.45.0-rc3
1.45.0-rc20
1.45.0-rc2
1.45.0-rc19
1.45.0-rc18
1.45.0-rc17
1.45.0-rc16
1.45.0-rc15
1.45.0-rc14
1.45.0-rc13
1.45.0-rc12
1.45.0-rc11
1.45.0-rc10
1.45.0-rc1
1.45.0-rc0
1.44.13
1.44.12
1.44.11
1.44.10
1.44.9
1.44.8
1.44.7
1.44.6
1.44.5
1.44.4
1.44.3
1.44.2
1.44.1
1.44.0
1.43.12
1.43.11
1.43.10
1.43.9
1.43.8
1.43.7
1.43.6
1.43.5
1.43.4
1.43.3
1.43.2
1.43.1
1.43.0
1.42.0
1.41.12
1.41.11
1.41.10
1.41.9
1.41.8
1.41.7
1.41.6
1.41.5
1.41.4
1.41.3
1.41.2
1.41.1
1.41.0
1.40.0
1.39.7
1.39.6
1.39.5
1.39.4
1.39.3
1.39.2
1.39.1
1.39.0
1.38.0
1.37.2
1.37.1
1.37.0
1.36.22
1.36.21
1.36.19
1.36.18
1.36.17
1.36.16
1.36.15
1.36.14
1.36.13
1.36.12
1.36.11
1.36.10
1.36.9
1.36.8
1.36.7
1.36.6
1.36.5
1.36.4
1.36.3
1.36.2
1.36.0
1.35.1
1.35.0
1.34.9
1.34.8
1.34.7
1.34.6
1.34.5
1.34.4
1.34.3
1.34.2
1.34.1
1.34.0
1.33.0
1.32.2
1.32.1
1.32.0
1.31.1
1.31.0
1.30.2
1.30.1
1.29.0-rc1
1.29.0-rc0
1.28.1
1.28.0
1.27.1
1.27.0
1.26.13
1.26.12
1.26.11
1.26.10
1.26.9
1.26.8
1.26.7
1.26.6
1.26.5
1.26.4
1.26.2
1.26.1
1.26.0
1.25.8
1.25.7
1.25.6
1.25.5
1.25.4
1.25.3
1.25.2
1.25.1
1.25.0
1.24.2
1.24.1
1.24.0
1.23.3
1.23.2
1.23.1
1.23.0
1.22.1
1.22.0
1.20.1
1.20.0
1.19.1
1.19.0
1.18.1
1.18.0
1.17.1
1.17.0
1.16.2
1.15.1
1.15.0
1.14.0
1.13.4
1.13.3
1.13.2
1.13.1
1.13.0
1.12.0
1.11.1
1.11.0
1.10.0
1.9.0
1.8.0
1.7.0
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.1
1.4.0
1.3.1
1.3.0
1.2.1
1.2.0
1.1.3
1.1.2
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
0.13.1
0.13.0
0.12.0
0.10.0
0.9.1
0.9.0
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.9
0.1.8
0.1.3
0.1.1
0.1.0
A declarative, extensible framework for building Elixir applications.
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash/engine/runner.ex
defmodule Ash.Engine.Runner do
@moduledoc false
defstruct [
:engine_pid,
:ref,
notified_of_complete?: false,
requests: [],
errors: [],
changeset: %{},
data: %{},
pid_info: %{},
dependencies: %{},
resource_notifications: [],
verbose?: false
]
alias Ash.Engine
alias Ash.Engine.Request
alias Ash.Error.Framework.SynchronousEngineStuck
require Logger
def run(requests, verbose?, ref, engine_pid \\ nil, pid_info \\ %{}) do
changeset =
Enum.find_value(requests, fn request ->
if request.manage_changeset? && not match?(%Request.UnresolvedField{}, request.changeset) do
request.changeset
end
end)
state = %__MODULE__{
requests: requests,
verbose?: verbose?,
engine_pid: engine_pid,
pid_info: pid_info,
ref: ref,
changeset: changeset,
resource_notifications: []
}
log(state, fn ->
"Synchronous engine starting - #{Enum.map_join(requests, ", ", & &1.name)}"
end)
new_state = run_to_completion(state)
new_state.requests
|> Enum.reduce(new_state, &add_data(&2, &1.path, &1.data))
|> Map.put(:resource_notifications, new_state.resource_notifications)
end
def run_to_completion(state) do
if Enum.all?(state.requests, &(&1.state in [:complete, :error])) do
# This allows for publishing any dependencies
new_state = run_iteration(state)
if new_state.engine_pid do
new_state =
if new_state.notified_of_complete? do
new_state
else
log(new_state, fn -> "notifying engine of local request completion" end)
GenServer.cast(new_state.engine_pid, :local_requests_complete)
%{new_state | notified_of_complete?: true}
end
wait_for_engine(new_state, true)
else
log(state, fn -> "Synchronous engine complete." end)
new_state
end
else
case run_iteration(state) do
new_state when new_state == state ->
if new_state.engine_pid do
wait_for_engine(new_state, false)
else
if new_state.errors == [] do
log(state, fn -> "Synchronous engine stuck:\n\n#{stuck_report(state)}" end)
GenServer.cast(state.engine_pid, :log_stuck_report)
add_error(new_state, :__engine__, SynchronousEngineStuck.exception([]))
else
new_state
end
end
new_state ->
run_to_completion(new_state)
end
end
end
defp stuck_report(state) do
Enum.map_join(state.requests, "\n", fn request ->
if request.state in [:complete, :error] do
to_string(request.name) <> ": " <> "#{inspect(request.state)}"
else
case Request.next(request) do
{:wait, _, _, []} ->
to_string(request.name) <> ": Waiting on nothing in state #{inspect(request.state)}"
{:wait, _, _, dependencies} ->
to_string(request.name) <> ": Waiting on #{dependency_names(dependencies, state)}"
_other ->
to_string(request.name) <> ": Not waiting, not complete"
end
end
end)
end
defp dependency_names(dependencies, state) do
Enum.map_join(dependencies, ", ", fn dependency ->
path = :lists.droplast(dependency)
field = List.last(dependency)
request = Enum.find(state.requests, &(&1.path == path))
request.name <> ": " <> to_string(field)
end)
end
defp wait_for_engine(state, complete?) do
log(state, fn -> "waiting for engine" end)
do_wait_for_engine(state, complete?)
end
defp do_wait_for_engine(%{engine_pid: engine_pid, ref: ref} = state, complete?) do
receive do
{^ref, {:update_changeset, changeset}} ->
run_to_completion(%{state | changeset: changeset})
{^ref, {:wont_receive, receiver_path, path, field}} ->
request = Enum.find(state.requests, &(&1.path == receiver_path))
new_state =
case Request.wont_receive(request, path, field) do
{:stop, :dependency_failed, new_request} ->
notify_error(:dependency_failed, state)
replace_request(state, %{new_request | state: :error})
end
run_to_completion(new_state)
{^ref, {:send_field, receiver_path, pid, dep}} ->
log(state, fn -> "notifying #{inspect(receiver_path)} of #{inspect(dep)}" end)
path = :lists.droplast(dep)
field = List.last(dep)
request = Enum.find(state.requests, &(&1.path == path))
new_state =
case Request.send_field(request, receiver_path, field) do
{:waiting, new_request, notifications, dependencies} ->
new_dependencies = build_dependencies(new_request, dependencies)
{new_state, new_notifications} =
state
|> replace_request(new_request)
|> store_dependencies(new_dependencies)
notify(new_state, notifications ++ new_notifications)
{:ok, new_request, notifications} ->
state
|> replace_request(new_request)
|> notify(notifications)
{:error, error, new_request} ->
if pid == self() do
send(self(), {ref, {:wont_receive, receiver_path, new_request.path, field}})
else
GenServer.cast(pid, {:wont_receive, receiver_path, new_request.path, field})
end
state
|> add_error(new_request.path, error)
|> replace_request(%{new_request | state: :error})
end
run_to_completion(new_state)
{^ref, {:field_value, receiver_path, request_path, field, value}} ->
receiver_path =
case receiver_path do
{_, path} -> path
path -> path
end
request = Enum.find(state.requests, &(&1.path == receiver_path))
case Request.receive_field(request, request_path, field, value) do
{:continue, new_request} ->
state
|> replace_request(new_request)
|> run_to_completion()
end
{^ref, {:notification, resource_notification}} ->
state
|> add_resource_notification(resource_notification)
|> run_to_completion()
{^ref, {:data, path, data}} ->
state
|> add_data(path, data)
|> run_to_completion()
{:DOWN, _, _, ^engine_pid,
{:shutdown, %{errored_requests: [], runner_ref: ^ref} = engine_state}} ->
log(state, fn -> "Engine complete" end)
handle_completion(state, engine_state, complete?, false)
{:DOWN, _, _, ^engine_pid, {:shutdown, %{runner_ref: ^ref} = engine_state}} ->
log(state, fn -> "Engine complete" end)
handle_completion(state, engine_state, complete?, true)
end
end
defp handle_completion(state, engine_state, complete?, engine_error?) do
new_state =
if complete? do
if engine_error? do
log(engine_state, fn -> "Engine shutdown error" end)
else
log(engine_state, fn -> "Engine complete, graceful shutdown" end)
end
state
else
run_to_completion(state)
end
add_engine_state(new_state, engine_state)
end
defp add_engine_state(state, engine_state) do
new_state = %{state | errors: engine_state.errors ++ state.errors}
flush(new_state)
end
defp flush(%{ref: ref} = state) do
receive do
{^ref, _} ->
flush(state)
after
0 ->
state
end
end
defp build_dependencies(request, dependencies) do
Enum.map(dependencies, fn dep ->
{request.path, dep}
end)
end
defp run_iteration(state) do
{new_state, notifications, dependencies} =
Enum.reduce(state.requests, {state, [], []}, fn request,
{state, notifications, dependencies} ->
{new_state, new_notifications, new_dependencies} = fully_advance_request(state, request)
{new_state, notifications ++ new_notifications, new_dependencies ++ dependencies}
end)
store_dependencies(new_state, dependencies, notifications)
end
defp store_dependencies(state, dependencies, notifications \\ []) do
log(state, fn -> "Storing generated dependencies" end)
{state, notifications, more_dependencies} =
dependencies
|> Enum.uniq()
|> Enum.reduce({state, notifications, []}, fn {request_path, dep},
{state, notifications, dependencies} ->
request = Enum.find(state.requests, &(&1.path == request_path))
path = :lists.droplast(dep)
field = List.last(dep)
case Enum.find(state.requests, &(&1.path == path)) do
nil ->
pid = Map.get(state.pid_info, path)
GenServer.cast(pid, {:send_field, request_path, self(), dep})
{state, notifications, dependencies}
depended_on_request ->
if request.error? || depended_on_request.error? do
{state, notifications, dependencies}
else
notify_local_request(
state,
notifications,
dependencies,
depended_on_request,
request,
field
)
end
end
end)
case more_dependencies do
[] ->
notify(state, notifications)
more_dependencies ->
store_dependencies(state, more_dependencies, notifications)
end
end
defp notify_local_request(
state,
notifications,
dependencies,
depended_on_request,
request,
field
) do
case Request.send_field(depended_on_request, request.path, field) do
{:ok, new_request, new_notifications} ->
{replace_request(state, new_request), notifications ++ new_notifications, dependencies}
{:waiting, new_request, new_notifications, new_dependencies} ->
new_dependencies = build_dependencies(new_request, new_dependencies)
new_state = replace_request(state, new_request)
{new_state, notifications ++ new_notifications, new_dependencies ++ dependencies}
{:error, error, new_request} ->
notify_error(error, state)
new_state =
state
|> replace_request(%{new_request | state: :error, error?: true})
|> add_error(new_request.path, error)
|> replace_request(%{new_request | state: :error, error?: true})
{new_state, notifications, dependencies}
end
end
defp notify_error(error, state) do
GenServer.cast(state.engine_pid, {:local_requests_failed, error})
end
defp notify(state, notifications) do
log(state, fn -> "sending/updating requests with notifications" end)
notifications
|> List.wrap()
|> Enum.uniq()
|> Enum.reduce(state, fn
{:set_extra_data, key, value}, state ->
%{state | data: Map.put(state.data, key, value)}
{:update_changeset, value}, state ->
%{state | changeset: value}
%Ash.Notifier.Notification{} = resource_notification, state ->
add_resource_notification(state, resource_notification)
{receiver_path, request_path, field, value}, state ->
case Enum.find(state.requests, &(&1.path == receiver_path)) do
nil ->
pid = Map.get(state.pid_info, receiver_path)
GenServer.cast(pid, {:field_value, receiver_path, request_path, field, value})
state
receiver_request ->
{:continue, new_request} =
Request.receive_field(receiver_request, request_path, field, value)
replace_request(state, new_request)
end
end)
end
defp replace_request(state, request) do
%{
state
| requests:
Enum.map(state.requests, fn existing_request ->
if existing_request.id == request.id do
request
else
existing_request
end
end)
}
end
defp fully_advance_request(state, request) do
case advance_request(request) do
{:ok, new_request, notifications, dependencies, resource_notification} ->
new_state =
state
|> replace_request(new_request)
|> add_resource_notification(resource_notification)
new_dependencies = build_dependencies(new_request, dependencies)
{new_state, notifications, new_dependencies}
{:ok, new_request, notifications, dependencies} ->
new_state = replace_request(state, new_request)
new_dependencies = build_dependencies(new_request, dependencies)
{new_state, notifications, new_dependencies}
{:error, error, notifications, new_request} ->
notify_error(error, state)
new_state =
state
|> add_error(new_request.path, error)
|> replace_request(%{new_request | state: :error})
{new_state, notifications, []}
{:error, error, new_request} ->
notify_error(error, state)
new_state =
state
|> add_error(new_request.path, error)
|> replace_request(%{new_request | state: :error})
{new_state, [], []}
end
end
defp add_resource_notification(state, resource_notification) do
if Ash.DataLayer.in_transaction?(resource_notification.resource) do
%{state | resource_notifications: [resource_notification | state.resource_notifications]}
else
Ash.Notifier.notify(resource_notification)
state
end
end
defp advance_request(%{state: :error} = request) do
{:ok, request, [], []}
end
defp advance_request(request) do
case Request.next(request) do
{:already_complete, new_request, new_notifications, new_dependencies} ->
{:ok, new_request, new_notifications, new_dependencies}
{:complete, new_request, new_notifications, new_dependencies} ->
if new_request.notify? do
resource_notification = Request.resource_notification(new_request)
{:ok, new_request, new_notifications, new_dependencies, resource_notification}
else
{:ok, new_request, new_notifications, new_dependencies}
end
{:continue, new_request, new_notifications} ->
{:ok, new_request, new_notifications, []}
{:error, error, notifications, new_request} ->
{:error, error, notifications, new_request}
{:error, error, new_request} ->
{:error, error, new_request}
{:wait, new_request, new_notifications, new_dependencies} ->
{:ok, new_request, new_notifications, new_dependencies}
end
end
defp add_data(state, path, data) do
%{
state
| data: Engine.put_nested_key(state.data, path, data)
}
end
defp add_error(state, path, errors) when is_list(errors) do
Enum.reduce(errors, state, &add_error(&2, path, &1))
end
defp add_error(state, path, error) do
path = List.wrap(path)
error = Ash.Error.to_ash_error(error)
error = Map.put(error, :path, path)
if error in state.errors do
state
else
%{state | errors: [error | state.errors]}
end
end
defp log(request, message, level \\ :debug)
defp log(%{verbose?: true}, message, level) do
Logger.log(level, fn -> ["Runner: ", message.()] end)
end
defp log(_, _, _) do
false
end
end