Packages
riptide
0.2.0
0.5.2
0.5.1
0.5.0-beta9
0.5.0-beta8
0.5.0-beta7
0.5.0-beta6
0.5.0-beta5
0.5.0-beta4
0.5.0-beta3
0.5.0-beta2
0.5.0-beta11
0.5.0-beta10
0.5.0-beta
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.13
0.3.12
0.3.11
0.3.10
0.3.9
0.3.8
0.3.7
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.3.0
0.3.0-bd63a38
0.2.79
0.2.78
0.2.74
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.15
0.1.14
0.1.13
0.1.12
0.1.11
0.1.10
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
0.1.0
A data first framework for building realtime applications
Current section
Files
Jump to
Current section
Files
lib/riptide/scheduler/scheduler.ex
defmodule Riptide.Scheduler do
require Logger
use Supervisor
def start_link(opts) do
Supervisor.start_link(__MODULE__, opts)
end
def init(_) do
Supervisor.init(
[
{Task.Supervisor, name: __MODULE__},
Riptide.Scheduler.Dispatch
],
strategy: :one_for_one
)
end
@root "riptide:scheduler"
def info(task), do: Riptide.query_path!([@root, task])
def stream(), do: Riptide.stream([@root])
def cancel(task) do
Riptide.Mutation.delete([@root, task])
end
def schedule_in(mod, fun, args, offset, key \\ nil),
do: schedule(:os.system_time(:millisecond) + offset, mod, fun, args, key)
def schedule(timestamp, mod, fun, args, key \\ nil) do
key = key || "SCH" <> Riptide.UUID.ascending()
Riptide.Mutation.merge([@root, key], %{
"key" => key,
"timestamp" => timestamp,
"mod" => mod,
"fun" => fun,
"args" => args,
"count" => 0
})
end
def execute(task) do
Logger.metadata(scheduler_task: task)
task
|> info()
|> case do
info = %{
"mod" => mod,
"args" => args,
"fun" => fun,
"timestamp" => timestamp
} ->
Logger.metadata(scheduler_mod: mod, scheduler_fun: fun)
mod = String.to_atom(mod)
fun = String.to_atom(fun)
try do
apply(mod, fun, args)
cond do
timestamp === Riptide.query_path!([@root, task, "timestamp"]) ->
Riptide.delete!([@root, task])
true ->
Riptide.merge!([@root, task, "count"], 0)
end
:ok
rescue
e ->
:error
|> Exception.format(e, __STACKTRACE__)
|> Logger.error(crash_reason: {e, __STACKTRACE__})
count = (info["count"] || 0) + 1
Riptide.merge!([@root, task, "count"], count)
Riptide.Retry.Basic
|> apply(:retry, [task, count])
|> case do
{:delay, amount} ->
:timer.sleep(amount)
execute(task)
:abort ->
Riptide.delete!([@root, task])
:abort
end
end
_ ->
{:error, :invalid_task}
end
end
end