Packages
ash_oban
0.1.9
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0-rc.1
0.8.0-rc.0
0.7.2
0.7.1
0.7.0
0.6.0
0.5.1
0.5.0
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.3.0
0.2.6
0.2.5
0.2.4
0.2.3
0.2.3-rc.1
0.2.3-rc.0
0.2.2
0.2.1
0.2.0
0.1.14
0.1.13
0.1.12
0.1.11
0.1.9
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
The extension for integrating Ash resources with Oban.
Current section
Files
Jump to
Current section
Files
lib/test.ex
defmodule AshOban.Test do
@moduledoc "Helpers for testing ash_oban triggers"
@type triggerable :: Ash.Resource.t() | {Ash.Resource.t(), atom()} | Ash.Api.t() | atom()
@doc """
Runs the schedulers for the given resource, api, or otp_app, or list of resources, apis, or otp_apps.
If the input is:
* a list - each item is passed into `schedule_and_run_triggers/1`, and the results are merged together.
* an otp_app - each api configured in the `ash_apis` of that otp_app is passed into `schedule_and_run_triggers/1`, and the results are merged together.
* an api - each reosurce configured in that api is passed into `schedule_and_run_triggers/1`, and the results are merged together.
* a tuple of {resource, :trigger_name} - that trigger is scheduled, and the results are merged together.
* a resource - each trigger configured in that resource is scheduled, and the results are merged together.
"""
@spec schedule_and_run_triggers(triggerable | list(triggerable)) :: %{
discard: non_neg_integer(),
cancelled: non_neg_integer(),
success: non_neg_integer(),
failure: non_neg_integer(),
snoozed: non_neg_integer()
}
def schedule_and_run_triggers(resources_or_apis_or_otp_apps)
when is_list(resources_or_apis_or_otp_apps) do
Enum.reduce(resources_or_apis_or_otp_apps, %{}, fn item, acc ->
item
|> schedule_and_run_triggers()
|> Map.merge(acc, fn _key, left, right ->
left + right
end)
end)
end
def schedule_and_run_triggers({resource, trigger_name}) do
triggers =
resource
|> AshOban.Info.oban_triggers()
|> Enum.filter(fn trigger ->
trigger.scheduler && trigger.name == trigger_name
end)
Enum.each(triggers, fn trigger ->
AshOban.schedule(resource, 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)
end
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)
Spark.Dsl.is?(resource_or_api_or_otp_app, Ash.Resource) ->
triggers =
resource_or_api_or_otp_app
|> AshOban.Info.oban_triggers()
|> Enum.filter(fn trigger ->
trigger.scheduler
end)
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