mirror of
https://github.com/ash-project/ash_oban.git
synced 2024-09-19 21:03:00 +12:00
improvement: support providing an otp app to schedule and run triggers
This commit is contained in:
parent
99a97b379d
commit
06c340569f
1 changed files with 42 additions and 45 deletions
87
lib/test.ex
87
lib/test.ex
|
@ -1,55 +1,52 @@
|
|||
defmodule AshOban.Test do
|
||||
@moduledoc "Helpers for testing ash_oban triggers"
|
||||
|
||||
@mix_app Mix.Project.config()[:app]
|
||||
|
||||
def schedule_and_run_triggers() do
|
||||
@mix_app
|
||||
|> IO.inspect()
|
||||
|> Application.get_env(:ash_apis, [])
|
||||
|> IO.inspect()
|
||||
|> List.wrap()
|
||||
|> Enum.reduce(%{}, fn api, acc ->
|
||||
api
|
||||
|> schedule_and_run_triggers()
|
||||
|> Map.merge(acc, fn _key, left, right ->
|
||||
left + right
|
||||
end)
|
||||
end)
|
||||
end
|
||||
|
||||
def schedule_and_run_triggers(resource_or_api) do
|
||||
if Spark.Dsl.is?(resource_or_api, Ash.Api) do
|
||||
resource_or_api
|
||||
|> Ash.Api.Info.resources()
|
||||
|> Enum.reduce(%{}, fn resource, acc ->
|
||||
resource
|
||||
|> schedule_and_run_triggers()
|
||||
|> Map.merge(acc, fn _key, left, right ->
|
||||
left + right
|
||||
def schedule_and_run_triggers(resource_or_api_or_otp_app) do
|
||||
cond do
|
||||
Spark.Dsl.is?(resource_or_api_or_otp_app, Ash.Api) ->
|
||||
resource_or_api_or_otp_app
|
||||
|> Ash.Api.Info.resources()
|
||||
|> Enum.reduce(%{}, fn resource, acc ->
|
||||
resource
|
||||
|> schedule_and_run_triggers()
|
||||
|> Map.merge(acc, fn _key, left, right ->
|
||||
left + right
|
||||
end)
|
||||
end)
|
||||
end)
|
||||
else
|
||||
triggers =
|
||||
AshOban.Info.oban_triggers(resource_or_api)
|
||||
|
||||
Enum.each(triggers, fn trigger ->
|
||||
AshOban.schedule(resource_or_api, trigger)
|
||||
end)
|
||||
Spark.Dsl.is?(resource_or_api_or_otp_app, Ash.Resource) ->
|
||||
triggers =
|
||||
AshOban.Info.oban_triggers(resource_or_api_or_otp_app)
|
||||
|
||||
queues =
|
||||
triggers
|
||||
|> Enum.map(& &1.queue)
|
||||
|> Enum.uniq()
|
||||
|
||||
# we drain each queue twice to do schedulers and then workers
|
||||
Enum.reduce(queues ++ queues, %{}, fn queue, acc ->
|
||||
[queue: queue]
|
||||
|> Oban.drain_queue()
|
||||
|> Map.merge(acc, fn _key, left, right ->
|
||||
left + right
|
||||
Enum.each(triggers, fn trigger ->
|
||||
AshOban.schedule(resource_or_api_or_otp_app, trigger)
|
||||
end)
|
||||
|
||||
queues =
|
||||
triggers
|
||||
|> Enum.map(& &1.queue)
|
||||
|> Enum.uniq()
|
||||
|
||||
# we drain each queue twice to do schedulers and then workers
|
||||
Enum.reduce(queues ++ queues, %{}, fn queue, acc ->
|
||||
[queue: queue]
|
||||
|> Oban.drain_queue()
|
||||
|> Map.merge(acc, fn _key, left, right ->
|
||||
left + right
|
||||
end)
|
||||
end)
|
||||
|
||||
true ->
|
||||
resource_or_api_or_otp_app
|
||||
|> Application.get_env(:ash_apis, [])
|
||||
|> List.wrap()
|
||||
|> Enum.reduce(%{}, fn api, acc ->
|
||||
api
|
||||
|> schedule_and_run_triggers()
|
||||
|> Map.merge(acc, fn _key, left, right ->
|
||||
left + right
|
||||
end)
|
||||
end)
|
||||
end)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
Loading…
Reference in a new issue