diff --git a/drift_js/CHANGELOG b/drift_js/CHANGELOG index b0fa781..6263979 100644 --- a/drift_js/CHANGELOG +++ b/drift_js/CHANGELOG @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Added +- Adds the `send_after` function to `runtime`, which allows sending messages + with a delay, or scheduling them on the task queue with a delay of zero. + ## [1.0.1] - 2025-17-20 ### Fixed diff --git a/drift_js/src/drift/js/channel.gleam b/drift_js/src/drift/js/channel.gleam index 17dd507..f44d1b6 100644 --- a/drift_js/src/drift/js/channel.gleam +++ b/drift_js/src/drift/js/channel.gleam @@ -2,9 +2,8 @@ //// Provided for convenience, not required to use `drift_js`. import gleam/javascript/promise.{type Promise} -import gleam/result -/// An unbounded single-consumer channel with synchronous sending +/// An unbounded single-consumer channel with non-blocking sending /// and asynchronous receiving. /// Intended to be used similarly to a `gleam/erlang` `Subject`. pub type Channel(a) diff --git a/drift_js/src/drift/js/internal/event_loop.gleam b/drift_js/src/drift/js/internal/event_loop.gleam index 7ee6fc9..a432ac9 100644 --- a/drift_js/src/drift/js/internal/event_loop.gleam +++ b/drift_js/src/drift/js/internal/event_loop.gleam @@ -44,6 +44,10 @@ pub fn error_if_stopped( @external(javascript, "../../../drift_event_loop.mjs", "send") pub fn send(loop: EventLoop(i), input: i) -> Nil +/// Uses `setTimeout` to call `send` after a delay. +@external(javascript, "../../../drift_event_loop.mjs", "send_after") +pub fn send_after(loop: EventLoop(i), delay: Int, input: i) -> Nil + /// Sets the time to the next time `receive` should return `Tick`. /// Only one timeout can be set at a time. /// Returns an error if a timeout is already set. diff --git a/drift_js/src/drift/js/runtime.gleam b/drift_js/src/drift/js/runtime.gleam index 911a4c6..83eccd7 100644 --- a/drift_js/src/drift/js/runtime.gleam +++ b/drift_js/src/drift/js/runtime.gleam @@ -35,10 +35,20 @@ pub type TerminalResult(a, e) { } /// Sends an input to be handled by the runtime. +/// This will resolve a promise under the hood, and thus completing the receive +/// will be scheduled as a microtask. pub fn send(runtime: Runtime(i), input: i) -> Nil { event_loop.send(runtime.loop, input) } +/// Sends an input to be handled by the runtime after a delay (in milliseconds). +/// Since triggering the receive from `send` will be scheduled as a microtask, +/// using `send_after` with a delay of 0 can be used to handle an input as +/// a task instead. +pub fn send_after(runtime: Runtime(i), delay: Int, input: i) -> Nil { + event_loop.send_after(runtime.loop, delay, input) +} + /// Similar to `process.call_forever` on Gleam on Erlang. pub fn call_forever( runtime: Runtime(i), diff --git a/drift_js/src/drift_event_loop.mjs b/drift_js/src/drift_event_loop.mjs index 20608a9..19a9ba9 100644 --- a/drift_js/src/drift_event_loop.mjs +++ b/drift_js/src/drift_event_loop.mjs @@ -38,6 +38,10 @@ export function send(loop, message) { return loop.send(new HandleInput(message)); } +export function send_after(loop, delay, message) { + setTimeout(() => send(loop, message), delay); +} + export function set_timeout(loop, after) { return loop.setTimeout(after); } diff --git a/drift_js/test/drift/js/internal/event_loop_test.gleam b/drift_js/test/drift/js/internal/event_loop_test.gleam index eacffa8..0638f7c 100644 --- a/drift_js/test/drift/js/internal/event_loop_test.gleam +++ b/drift_js/test/drift/js/internal/event_loop_test.gleam @@ -2,6 +2,7 @@ import drift/js/internal/event_loop.{ type Event, type EventLoop, AlreadyReceiving, AlreadyTicking, HandleInput, Stopped, Tick, } +import drift/js/runtime import gleam/javascript/promise.{type Promise, await} pub fn receive_when_empty_test() { @@ -84,6 +85,24 @@ pub fn send_cancels_timeout_test() { promise.resolve(Nil) } +pub fn send_after_test() { + let loop = event_loop.start() + let start = runtime.now() + + event_loop.send_after(loop, 10, 42) + + let assert Ok(result) = event_loop.receive(loop) + use result <- await(timeout(result, 20)) + assert result == Ok(HandleInput(42)) + + // Assert reasonably accurate delay + let elapsed = runtime.now() - start + assert elapsed >= 10 + assert elapsed <= 12 + + promise.resolve(Nil) +} + pub fn queued_receive_cancels_timeout_test() { let loop = event_loop.start() diff --git a/drift_js/test/drift/shceduling_test.gleam b/drift_js/test/drift/shceduling_test.gleam new file mode 100644 index 0000000..d864ec0 --- /dev/null +++ b/drift_js/test/drift/shceduling_test.gleam @@ -0,0 +1,66 @@ +import drift +import drift/js/channel.{type Channel} +import drift/js/runtime +import gleam/javascript/promise.{type Promise} +import gleam/list + +type State { + State(value: Int, remaining_sends: Int) +} + +// Checks that a single runtime that sends messages to itself repeatedly +// still allows other runtimes to run. +pub fn multiple_runtime_scheduling_test() -> Promise(Nil) { + let results = channel.new() + let #(termination1, runtime1) = + runtime.start(State(1, 5), fn(_) { results }, handle_input, handle_output) + let #(termination2, runtime2) = + runtime.start(State(2, 5), fn(_) { results }, handle_input, handle_output) + + // Kick off the event handling + runtime.send(runtime1, Nil) + runtime.send(runtime2, Nil) + + // Wait for termination + use _ <- promise.await(termination1) + use _ <- promise.await(termination2) + + assert channel_to_list(results, []) == [1, 2, 1, 2, 1, 2, 1, 2, 1, 2] + + promise.resolve(Nil) +} + +fn handle_output( + context: drift.EffectContext(channel.Channel(Int)), + value: Int, + send: fn(Nil) -> Nil, +) -> Result(drift.EffectContext(channel.Channel(Int)), Nil) { + // Send who was scheduled to results + let results = drift.read_effect_context(context) + channel.send(results, value) + + // Trigger another input to be handled + send(Nil) + Ok(context) +} + +fn handle_input( + ctx: drift.Context(a, Int), + state: State, + _, +) -> drift.Step(State, a, Int, c) { + case state.remaining_sends { + 0 -> drift.stop(ctx, state) + remaining_sends -> + ctx + |> drift.output(state.value) + |> drift.continue(State(..state, remaining_sends: remaining_sends - 1)) + } +} + +fn channel_to_list(channel: Channel(a), values: List(a)) -> List(a) { + case channel.try_receive(channel) { + Ok(value) -> channel_to_list(channel, [value, ..values]) + Error(Nil) -> list.reverse(values) + } +} diff --git a/drift_js/test/drift_js_test.gleam b/drift_js/test/drift_js_test.gleam index 5ca126e..6c68a0d 100644 --- a/drift_js/test/drift_js_test.gleam +++ b/drift_js/test/drift_js_test.gleam @@ -1,9 +1,10 @@ import drift.{type Action, type Context, type Effect, type Step} import drift/js/runtime.{ - type Runtime, type TerminalResult, CallTimedOut, RuntimeStopped, + type Runtime, type TerminalResult, CallTimedOut, RuntimeStopped, Terminated, } import exemplify import gleam/javascript/promise.{type Promise, await} +import gleam/list import gleeunit pub fn main() -> Nil { @@ -112,6 +113,32 @@ pub fn call_timeout_test() { promise.resolve(Nil) } +pub fn send_after_0_is_delayed_test() { + use <- timeout(100) + + let #(result, rt) = + start_without_io([], fn(ctx, state, input) { + let state = [input, ..state] + case list.length(state) { + 4 -> drift.stop(ctx, state) + _ -> drift.continue(ctx, state) + } + }) + + runtime.send_after(rt, 0, "after1") + runtime.send(rt, "immediate1") + runtime.send_after(rt, 0, "after2") + runtime.send(rt, "immediate2") + use result <- promise.await(result) + + // Reverse list for clearer assertion + let assert Terminated(result) = result + let result = list.reverse(result) + assert result == ["immediate1", "immediate2", "after1", "after2"] + + promise.resolve(Nil) +} + fn noop(ctx: Context(i, o), state: s, _: i) -> Step(s, i, o, e) { ctx |> drift.continue(state) } @@ -133,7 +160,7 @@ fn stop(ctx: Context(Bool, o), state: s, stop: Bool) -> Step(s, Bool, o, e) { } } -pub fn start_with_action_executor( +fn start_with_action_executor( state: s, next: fn(Context(i, Action(a)), s, i) -> Step(s, i, Action(a), e), ) -> #(Promise(TerminalResult(s, e)), Runtime(i)) {