Packages

A scheduling expansion for the Clockwork library, providing a way to define and manage recurring tasks.

Current section

Files

Jump to
clockwork_schedule src clockwork_schedule.erl
Raw

src/clockwork_schedule.erl

-module(clockwork_schedule).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/clockwork_schedule.gleam").
-export([new/3, with_logging/1, with_time_offset/2, stop/1, start/2, supervised/2]).
-export_type([message/0, state/0, scheduler/0]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
?MODULEDOC(
"////////////////////////////////////////////////////////////////////////////////\n"
"////////////////////////////////////////////////////////////////////////////////\n"
).
-type message() :: run | stop.
-type state() :: {state,
binary(),
gleam@erlang@process:subject(message()),
clockwork:cron(),
fun(() -> nil),
gleam@time@duration:duration()}.
-opaque scheduler() :: {scheduler,
binary(),
clockwork:cron(),
fun(() -> nil),
boolean(),
gleam@time@duration:duration()}.
-file("src/clockwork_schedule.gleam", 112).
?DOC(
" Creates a new scheduler configuration.\n"
" \n"
" ## Parameters\n"
" \n"
" - `id`: A unique identifier for this scheduler (used in logging)\n"
" - `cron`: A cron expression defining when the job should run\n"
" - `job`: The function to execute on each scheduled occurrence\n"
" \n"
" ## Example\n"
" \n"
" ```gleam\n"
" import clockwork\n"
" \n"
" let assert Ok(cron) = clockwork.from_string(\"0 */2 * * *\") // Every 2 hours\n"
" let scheduler = clockwork_schedule.new(\"data_sync\", cron, fn() {\n"
" database.sync_remote_data()\n"
" })\n"
" ```\n"
).
-spec new(binary(), clockwork:cron(), fun(() -> nil)) -> scheduler().
new(Id, Cron, Job) ->
{scheduler, Id, Cron, Job, false, {duration, 0, 0}}.
-file("src/clockwork_schedule.gleam", 131).
?DOC(
" Enables logging for the scheduler.\n"
" \n"
" When logging is enabled, the scheduler will log:\n"
" - When a job starts executing (with timestamp)\n"
" - When the scheduler is stopped\n"
" \n"
" Logging uses the `logging` library and outputs at the `Info` level.\n"
" \n"
" ## Example\n"
" \n"
" ```gleam\n"
" let scheduler = \n"
" clockwork_schedule.new(\"cleanup\", cron, cleanup_fn)\n"
" |> clockwork_schedule.with_logging() // Enable logging\n"
" ```\n"
).
-spec with_logging(scheduler()) -> scheduler().
with_logging(Scheduler) ->
{scheduler,
erlang:element(2, Scheduler),
erlang:element(3, Scheduler),
erlang:element(4, Scheduler),
true,
erlang:element(6, Scheduler)}.
-file("src/clockwork_schedule.gleam", 171).
?DOC(
" Sets a time zone offset for the scheduler.\n"
" \n"
" By default, schedulers use the system's UTC offset. Use this function\n"
" to run scheduled tasks in a specific time zone.\n"
" \n"
" ## Parameters\n"
" \n"
" - `scheduler`: The scheduler to configure\n"
" - `offset`: The UTC offset as a Duration (positive for east, negative for west)\n"
" \n"
" ## Example\n"
"\n"
" ```gleam\n"
" import gleam/erlang/process\n"
" import gleam/time/duration\n"
"\n"
" // Configure for UTC+9 (Tokyo)\n"
" let tokyo_offset = duration.from_hours(9)\n"
"\n"
" let tokyo_scheduler =\n"
" clockwork_schedule.new(\"tokyo_job\", cron, job_fn)\n"
" |> clockwork_schedule.with_time_offset(tokyo_offset)\n"
"\n"
" let tokyo_name = process.new_name(\"tokyo_job\")\n"
" let assert Ok(_subject) = clockwork_schedule.start(tokyo_scheduler, tokyo_name)\n"
"\n"
" // Configure for UTC-5 (New York)\n"
" let ny_offset = duration.from_hours(-5)\n"
"\n"
" let ny_scheduler =\n"
" clockwork_schedule.new(\"ny_job\", cron, job_fn)\n"
" |> clockwork_schedule.with_time_offset(ny_offset)\n"
"\n"
" let ny_name = process.new_name(\"ny_job\")\n"
" let assert Ok(_subject) = clockwork_schedule.start(ny_scheduler, ny_name)\n"
" ```\n"
).
-spec with_time_offset(scheduler(), gleam@time@duration:duration()) -> scheduler().
with_time_offset(Scheduler, Offset) ->
{scheduler,
erlang:element(2, Scheduler),
erlang:element(3, Scheduler),
erlang:element(4, Scheduler),
erlang:element(5, Scheduler),
Offset}.
-file("src/clockwork_schedule.gleam", 366).
?DOC(
" Gracefully stops a running scheduler.\n"
"\n"
" Sends a stop message to the scheduler, which will:\n"
" 1. Cancel any pending job executions\n"
" 2. Log a stop message (if logging is enabled)\n"
" 3. Terminate the scheduler actor\n"
"\n"
" ## Parameters\n"
"\n"
" - `name`: The name of the scheduler process to stop\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" import gleam/erlang/process\n"
"\n"
" let name = process.new_name(\"my_scheduler\")\n"
" let assert Ok(_subject) = clockwork_schedule.start(scheduler, name)\n"
"\n"
" // Run for some time...\n"
" process.sleep(60_000) // 1 minute\n"
"\n"
" // Gracefully stop\n"
" clockwork_schedule.stop(name)\n"
" ```\n"
"\n"
" ## Note\n"
"\n"
" After calling `stop`, the scheduler will terminate. To restart scheduling,\n"
" create and start a new scheduler with a new name.\n"
).
-spec stop(gleam@erlang@process:name(message())) -> nil.
stop(Name) ->
_pipe = gleam@erlang@process:named_subject(Name),
gleam@erlang@process:send(_pipe, stop).
-file("src/clockwork_schedule.gleam", 395).
-spec enqueue_job(clockwork:cron(), state()) -> gleam@erlang@process:timer().
enqueue_job(Cron, State) ->
Now = gleam@time@timestamp:system_time(),
Next_occurrence = begin
_pipe = clockwork:next_occurrence(Cron, Now, erlang:element(6, State)),
_pipe@1 = gleam@time@timestamp:difference(Now, _pipe),
_pipe@2 = gleam@time@duration:to_seconds_and_nanoseconds(_pipe@1),
(fun(Tuple) ->
{Seconds, Nanoseconds} = Tuple,
Milliseconds = (Seconds * 1000) + (Nanoseconds div 1000000),
case Milliseconds < 0 of
true ->
100;
false ->
Milliseconds
end
end)(_pipe@2)
end,
gleam@erlang@process:send_after(
erlang:element(3, State),
Next_occurrence,
run
).
-file("src/clockwork_schedule.gleam", 371).
-spec loop(state(), message()) -> gleam@otp@actor:next(state(), any()).
loop(State, Message) ->
case Message of
run ->
logging:log(
info,
<<<<<<"[CLOCKWORK] Running job: "/utf8,
(erlang:element(2, State))/binary>>/binary,
" at "/utf8>>/binary,
(begin
_pipe = gleam@time@timestamp:system_time(),
_pipe@1 = gleam@time@timestamp:add(
_pipe,
erlang:element(6, State)
),
_pipe@2 = gleam@time@timestamp:to_unix_seconds(_pipe@1),
gleam_stdlib:float_to_string(_pipe@2)
end)/binary>>
),
proc_lib:spawn_link(erlang:element(5, State)),
enqueue_job(erlang:element(4, State), State),
gleam@otp@actor:continue(State);
stop ->
logging:log(
info,
<<"[CLOCKWORK] Stopping job: "/utf8,
(erlang:element(2, State))/binary>>
),
gleam@otp@actor:stop()
end.
-file("src/clockwork_schedule.gleam", 184).
-spec start_actor(scheduler(), gleam@erlang@process:name(message())) -> {ok,
gleam@otp@actor:started(gleam@erlang@process:subject(message()))} |
{error, gleam@otp@actor:start_error()}.
start_actor(Scheduler, Name) ->
case erlang:element(5, Scheduler) of
true ->
logging_ffi:configure();
false ->
nil
end,
_pipe@4 = gleam@otp@actor:new_with_initialiser(
100,
fun(Self) ->
State = {state,
erlang:element(2, Scheduler),
Self,
erlang:element(3, Scheduler),
erlang:element(4, Scheduler),
erlang:element(6, Scheduler)},
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
gleam@erlang@process:select(_pipe, Self)
end,
enqueue_job(erlang:element(3, Scheduler), State),
_pipe@1 = gleam@otp@actor:initialised(State),
_pipe@2 = gleam@otp@actor:selecting(_pipe@1, Selector),
_pipe@3 = gleam@otp@actor:returning(_pipe@2, Self),
{ok, _pipe@3}
end
),
_pipe@5 = gleam@otp@actor:on_message(_pipe@4, fun loop/2),
_pipe@6 = gleam@otp@actor:named(_pipe@5, Name),
gleam@otp@actor:start(_pipe@6).
-file("src/clockwork_schedule.gleam", 265).
?DOC(
" Starts an unsupervised scheduler.\n"
"\n"
" This function starts a scheduler as a standalone actor that will run\n"
" according to its cron expression. For production use, prefer `supervised`\n"
" to run the scheduler under OTP supervision for better fault tolerance.\n"
"\n"
" ## Parameters\n"
"\n"
" - `scheduler`: The scheduler configuration to start\n"
" - `name`: A unique name for the scheduler process\n"
"\n"
" ## Returns\n"
"\n"
" - `Ok(Subject(Message))`: A subject to send messages to the scheduler\n"
" - `Error(actor.StartError)`: If the scheduler fails to start\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" import clockwork\n"
" import clockwork_schedule\n"
" import gleam/erlang/process\n"
"\n"
" pub fn main() {\n"
" let assert Ok(cron) = clockwork.from_string(\"*/30 * * * *\") // Every 30 minutes\n"
"\n"
" let scheduler =\n"
" clockwork_schedule.new(\"metrics\", cron, fn() {\n"
" metrics.collect_and_report()\n"
" })\n"
" |> clockwork_schedule.with_logging()\n"
"\n"
" let name = process.new_name(\"metrics\")\n"
" let assert Ok(_subject) = clockwork_schedule.start(scheduler, name)\n"
"\n"
" // The scheduler is now running\n"
" // Stop it when done:\n"
" clockwork_schedule.stop(name)\n"
" }\n"
" ```\n"
"\n"
" ## Note\n"
"\n"
" The scheduler will continue running until explicitly stopped with `stop`\n"
" or until the process crashes. For automatic restart on failure, use\n"
" `supervised` instead.\n"
).
-spec start(scheduler(), gleam@erlang@process:name(message())) -> {ok,
gleam@otp@actor:started(gleam@erlang@process:subject(message()))} |
{error, gleam@otp@actor:start_error()}.
start(Scheduler, Name) ->
start_actor(Scheduler, Name).
-file("src/clockwork_schedule.gleam", 329).
?DOC(
" Creates a child specification for running the scheduler under OTP supervision.\n"
"\n"
" This is the recommended way to run schedulers in production. The scheduler\n"
" will be automatically restarted if it crashes, ensuring your scheduled\n"
" tasks remain reliable.\n"
"\n"
" ## Parameters\n"
"\n"
" - `scheduler`: The scheduler configuration\n"
" - `name`: A unique name for the scheduler process\n"
"\n"
" ## Returns\n"
"\n"
" A `supervision.ChildSpec` that can be added to your supervision tree.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" import clockwork\n"
" import clockwork_schedule\n"
" import gleam/erlang/process\n"
" import gleam/otp/static_supervisor as supervisor\n"
"\n"
" pub fn main() {\n"
" let assert Ok(cron) = clockwork.from_string(\"0 * * * *\") // Every hour\n"
"\n"
" let scheduler =\n"
" clockwork_schedule.new(\"hourly_task\", cron, fn() {\n"
" perform_hourly_maintenance()\n"
" })\n"
" |> clockwork_schedule.with_logging()\n"
"\n"
" // Create a unique name for the scheduler\n"
" let name = process.new_name(\"hourly_task\")\n"
"\n"
" // Create the child spec\n"
" let schedule_child_spec =\n"
" clockwork_schedule.supervised(scheduler, name)\n"
"\n"
" // Add to supervision tree\n"
" let assert Ok(_sup) =\n"
" supervisor.new()\n"
" |> supervisor.add(schedule_child_spec)\n"
" |> supervisor.start()\n"
"\n"
" // The scheduler is now running under supervision\n"
" // Control it using the name:\n"
" // clockwork_schedule.stop(name)\n"
" process.sleep_forever()\n"
" }\n"
" ```\n"
"\n"
" ## Fault Tolerance\n"
"\n"
" If the scheduler crashes, the supervisor will automatically restart it.\n"
" The new instance will recalculate the next occurrence and continue\n"
" scheduling jobs as expected.\n"
).
-spec supervised(scheduler(), gleam@erlang@process:name(message())) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(message())).
supervised(Scheduler, Name) ->
gleam@otp@supervision:worker(
fun() ->
gleam@result:'try'(
start_actor(Scheduler, Name),
fun(Started) -> {ok, Started} end
)
end
).