From 399e30d09eadf670d59d3123d82ce1d3c7a56827 Mon Sep 17 00:00:00 2001 From: Sakari Bergen Date: Sun, 20 Jul 2025 11:54:51 +0300 Subject: [PATCH 1/5] Fix missing whitespace --- drift/gleam.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/drift/gleam.toml b/drift/gleam.toml index ce4db18..6c6c5c3 100644 --- a/drift/gleam.toml +++ b/drift/gleam.toml @@ -1,7 +1,7 @@ name = "drift" version = "1.0.0" -description = """Define pure functional cores in target-agnostic Gleam,\ +description = """Define pure functional cores in target-agnostic Gleam, \ which can wrapped to run on both the Erlang and JS targets.""" licences = ["Apache-2.0"] repository = { type = "github", user = "sbergen", repo = "drift", path = "drift" } From 694fd07f21551c6b58b45c162cbf6890011a1994 Mon Sep 17 00:00:00 2001 From: Sakari Bergen Date: Sun, 20 Jul 2025 11:58:30 +0300 Subject: [PATCH 2/5] Revamp starting API to support named subjects --- drift_actor/README.md | 40 +++--- drift_actor/src/drift/actor.gleam | 118 +++++++++++------- drift_actor/test/drift/actor/echo_actor.gleam | 6 +- .../test/drift/actor/inputs_test.gleam | 7 +- .../test/drift/actor/supervision_test.gleam | 31 +++-- drift_actor/test/example.gleam | 40 +++--- 6 files changed, 144 insertions(+), 98 deletions(-) diff --git a/drift_actor/README.md b/drift_actor/README.md index 23b661b..0ee095f 100644 --- a/drift_actor/README.md +++ b/drift_actor/README.md @@ -14,7 +14,6 @@ gleam add drift_actor@1 import drift import drift/actor import gleam/erlang/process -import gleam/function import gleam/io import gleam/option.{None, Some} import gleam/string @@ -22,25 +21,34 @@ import gleam/string pub fn main() { // Start a stepper that adds all the numbers sent to it, // until None is encountered - let assert Ok(subject) = + let assert Ok(started) = actor.using_io( - // No inputs in this example - fn() { process.new_selector() }, - function.identity, - fn(ctx, output) { + // No external inputs in this example + with_initial_state: fn() { process.new_selector() }, + selecting_inputs: fn(selector) { selector }, + // Print all the outputs + handling_outputs_with: fn(io_state, output) { io.println(string.inspect(output)) - Ok(ctx) + Ok(io_state) }, ) - |> actor.start(100, 0, fn(ctx, state, input) { - case input { - Some(input) -> drift.continue(ctx, state + input) - None -> - ctx - |> drift.output(state) - |> drift.stop(state) - } - }) + // This defines the pure part: + |> actor.with_stepper( + with_initial_state: 0, + handling_inputs_with: fn(ctx, state, input) { + case input { + Some(input) -> drift.continue(ctx, state + input) + None -> + ctx + |> drift.output(state) + |> drift.stop(state) + } + }, + ) + // Start the actor without wrapping the subject + |> actor.start(100, fn(subject) { subject }) + + let subject = started.data process.send(subject, Some(40)) process.send(subject, Some(2)) diff --git a/drift_actor/src/drift/actor.gleam b/drift_actor/src/drift/actor.gleam index 52a9167..fe90c08 100644 --- a/drift_actor/src/drift/actor.gleam +++ b/drift_actor/src/drift/actor.gleam @@ -14,8 +14,7 @@ import gleam/result import gleam/string /// The state of the wrapping actor. -/// Should not be directly used, but must be public for supporting actor builders. -pub opaque type State(state, io, input, output, error) { +type State(state, io, input, output, error) { State( stepper: drift.Stepper(state, input), timer: Option(process.Timer), @@ -30,8 +29,7 @@ pub opaque type State(state, io, input, output, error) { } /// The message type of the wrapping actor. -/// Should not be directly used, but must be public for supporting actor builders. -pub opaque type Msg(i) { +type Msg(i) { Tick HandleInput(i) } @@ -53,88 +51,112 @@ pub opaque type IoDriver(state, input, output) { /// and returns the new effect context or an error. /// `get_selector` should extract the `Selector` for inputs from the io state. pub fn using_io( - init: fn() -> state, - get_selector: fn(state) -> Selector(input), - handle_output: fn(EffectContext(state), output) -> + with_initial_state init: fn() -> state, + selecting_inputs get_selector: fn(state) -> Selector(input), + handling_outputs_with handle_output: fn(EffectContext(state), output) -> Result(EffectContext(state), String), ) -> IoDriver(state, input, output) { IoDriver(init, get_selector, handle_output) } -/// Starts a simple actor linked to the current process. No supervision. -/// On success, messages sent to the returned `Subject` will be handled -/// by the given input handling function, and the actor will take care of -/// all state transitions. -pub fn start( - io_driver: IoDriver(io, i, o), - timeout: Int, - state: s, - handle_input: fn(Context(i, o), s, i) -> Step(s, i, o, e), -) -> Result(Subject(i), actor.StartError) { - builder(io_driver, timeout, state, handle_input, None, fn(_) { Nil }) - |> actor.start() - |> result.map(fn(init) { init.data }) +pub opaque type Builder(s, io, i, o, e) { + Builder( + io_driver: IoDriver(io, i, o), + initial_state: s, + handle_input: fn(Context(i, o), s, i) -> Step(s, i, o, e), + name: Option(process.Name(i)), + ) } -/// Behaves like `start`, except that it provides an actor builder that allows -/// more configurability for advanced use cases, like supervision. -/// NOTE: Do not reconfigure `on_message`! -/// -/// If provided, `report_to` will be sent a message returned by `make_report` -/// when the actor is initialized. -pub fn builder( +/// Configures the drift stepper to use with the actor, +/// this is the second mandatory step after configuring the IO, +/// after which other options are optional. +pub fn with_stepper( io_driver: IoDriver(io, i, o), + with_initial_state initial_state: s, + handling_inputs_with handle_input: fn(Context(i, o), s, i) -> Step(s, i, o, e), +) { + Builder(io_driver:, initial_state:, handle_input:, name: None) +} + +/// Configures the actor to use a named process and subject for the inputs. +/// This allows using supervision and not invalidating previous subjects +/// on restarts. +pub fn named( + builder: Builder(s, io, i, o, e), + name: process.Name(i), +) -> Builder(s, io, i, o, e) { + Builder(..builder, name: Some(name)) +} + +/// Starts the actor with the provided configuration, timeout and possible +/// extra initialization. +/// The extra initialization will happen as part of the actor initialization, +/// and contributes to the timeout limit. +/// If you want to do asynchronous initialization, consider sending a message +/// to another process. +pub fn start( + builder: Builder(s, io, i, o, e), timeout: Int, - state: s, - handle_input: fn(Context(i, o), s, i) -> Step(s, i, o, e), - report_to: Option(Subject(r)), - make_report: fn(Subject(i)) -> r, -) -> actor.Builder(State(s, io, i, o, e), Msg(i), Subject(i)) { + init: fn(Subject(i)) -> result, +) -> Result(actor.Started(result), actor.StartError) { actor.new_with_initialiser(timeout, fn(self) { - let io_state = io_driver.init() + use inputs <- result.try(create_input_subject(builder.name)) + let io_state = builder.io_driver.init() + let #(stepper, effect_ctx) = drift.new(builder.initial_state, io_state) - let #(stepper, effect_ctx) = drift.new(state, io_state) - - let inputs = process.new_subject() let base_selector = process.new_selector() |> process.select_map(inputs, HandleInput) |> process.select(self) - let input_selector = io_driver.get_selector(io_state) + let input_selector = builder.io_driver.get_selector(io_state) let state = State( stepper:, timer: None, effect_ctx:, - io_driver:, + io_driver: builder.io_driver, input_selector:, self:, base_selector:, - handle_input:, + handle_input: builder.handle_input, ) - let init = + let initialised = actor.initialised(state) |> actor.selecting( input_selector |> process.map_selector(HandleInput) |> process.merge_selector(base_selector), ) - |> actor.returning(inputs) - - case report_to { - Some(subject) -> process.send(subject, make_report(inputs)) - None -> Nil - } + |> actor.returning(init(inputs)) - Ok(init) + Ok(initialised) }) |> actor.on_message(handle_message) + |> actor.start() +} + +fn create_input_subject( + name: Option(process.Name(i)), +) -> Result(Subject(i), String) { + case name { + Some(name) -> { + use _ <- result.try( + process.register(process.self(), name) + |> result.replace_error("name already registered"), + ) + Ok(process.named_subject(name)) + } + None -> { + Ok(process.new_subject()) + } + } } -/// Similar to `process.call_forver`, but dispatches to the stepper. +/// Similar to `process.call_forever`, but dispatches to the stepper. pub fn call_forever( actor: Subject(message), sending make_request: fn(Effect(reply)) -> message, diff --git a/drift_actor/test/drift/actor/echo_actor.gleam b/drift_actor/test/drift/actor/echo_actor.gleam index 22b0a7a..21dab95 100644 --- a/drift_actor/test/drift/actor/echo_actor.gleam +++ b/drift_actor/test/drift/actor/echo_actor.gleam @@ -2,6 +2,7 @@ import drift.{type Action, type Effect} import drift/actor import gleam/dict.{type Dict} import gleam/erlang/process.{type Subject} +import gleam/function // Erlang part @@ -16,9 +17,10 @@ pub fn new() -> Subject(Input) { Ok(drift.perform_effect(ctx, action)) }, ) - |> actor.start(100, State(0, dict.new()), handle_input) + |> actor.with_stepper(State(0, dict.new()), handle_input) + |> actor.start(100, function.identity) - actor + actor.data } // Generic part diff --git a/drift_actor/test/drift/actor/inputs_test.gleam b/drift_actor/test/drift/actor/inputs_test.gleam index e0229a0..0806fdc 100644 --- a/drift_actor/test/drift/actor/inputs_test.gleam +++ b/drift_actor/test/drift/actor/inputs_test.gleam @@ -48,7 +48,7 @@ pub fn input_change_test() { fn start_actor(inputs: Subject(Subject(Input))) -> Subject(Input) { // Actor that sums inputs and can switch input subjects - let assert Ok(actor_subject) = + let assert Ok(started) = actor.using_io( // the io state is the input selector fn() { @@ -74,7 +74,7 @@ fn start_actor(inputs: Subject(Subject(Input))) -> Subject(Input) { }) }, ) - |> actor.start(100, 0, fn(ctx, state, input) { + |> actor.with_stepper(0, fn(ctx, state, input) { case input { Sum(value) -> ctx |> drift.continue(state + value) @@ -90,6 +90,7 @@ fn start_actor(inputs: Subject(Subject(Input))) -> Subject(Input) { } } }) + |> actor.start(100, function.identity) - actor_subject + started.data } diff --git a/drift_actor/test/drift/actor/supervision_test.gleam b/drift_actor/test/drift/actor/supervision_test.gleam index bb10196..3964336 100644 --- a/drift_actor/test/drift/actor/supervision_test.gleam +++ b/drift_actor/test/drift/actor/supervision_test.gleam @@ -2,17 +2,16 @@ import drift import drift/actor import gleam/erlang/process.{type Pid, type Subject} import gleam/function -import gleam/option.{Some} -import gleam/otp/actor as otp_actor import gleam/otp/static_supervisor import gleam/otp/supervision pub fn supervision_test() { let children = process.new_subject() + let child_name = process.new_name("supervision_test") let assert Ok(supervisor) = static_supervisor.new(static_supervisor.OneForOne) - |> static_supervisor.add(init_panicking_child(children)) + |> static_supervisor.add(init_panicking_child(children, child_name)) |> static_supervisor.start let assert Ok(#(pid1, inputs)) = process.receive(children, 10) @@ -24,14 +23,20 @@ pub fn supervision_test() { let assert Ok(#(pid2, _)) = process.receive(children, 10) as "Child should restart" - let assert False = process.is_alive(pid1) - let assert True = process.is_alive(pid2) + assert !process.is_alive(pid1) + assert process.is_alive(pid2) + + // We should be able to use the same named subject again, + // to kill the newly spawned child. + process.send(inputs, Nil) + assert !process.is_alive(pid2) process.send_exit(supervisor.pid) } fn init_panicking_child( subject: Subject(#(Pid, Subject(Nil))), + name: process.Name(Nil), ) -> supervision.ChildSpecification(Subject(Nil)) { supervision.worker(fn() { actor.using_io( @@ -40,13 +45,13 @@ fn init_panicking_child( function.identity, fn(ctx, _) { Ok(ctx) }, ) - |> actor.builder( - 100, - Nil, - fn(ctx, _, _) { drift.stop_with_error(ctx, "Stepper failing!") }, - Some(subject), - fn(inputs) { #(process.self(), inputs) }, - ) - |> otp_actor.start + |> actor.with_stepper(Nil, fn(ctx, _, _) { + drift.stop_with_error(ctx, "Stepper failing!") + }) + |> actor.named(name) + |> actor.start(100, fn(inputs) { + process.send(subject, #(process.self(), inputs)) + inputs + }) }) } diff --git a/drift_actor/test/example.gleam b/drift_actor/test/example.gleam index 00992db..3584aaa 100644 --- a/drift_actor/test/example.gleam +++ b/drift_actor/test/example.gleam @@ -1,7 +1,6 @@ import drift import drift/actor import gleam/erlang/process -import gleam/function import gleam/io import gleam/option.{None, Some} import gleam/string @@ -9,25 +8,34 @@ import gleam/string pub fn main() { // Start a stepper that adds all the numbers sent to it, // until None is encountered - let assert Ok(subject) = + let assert Ok(started) = actor.using_io( - // No inputs in this example - fn() { process.new_selector() }, - function.identity, - fn(ctx, output) { + // No external inputs in this example + with_initial_state: fn() { process.new_selector() }, + selecting_inputs: fn(selector) { selector }, + // Print all the outputs + handling_outputs_with: fn(io_state, output) { io.println(string.inspect(output)) - Ok(ctx) + Ok(io_state) }, ) - |> actor.start(100, 0, fn(ctx, state, input) { - case input { - Some(input) -> drift.continue(ctx, state + input) - None -> - ctx - |> drift.output(state) - |> drift.stop(state) - } - }) + // This defines the pure part: + |> actor.with_stepper( + with_initial_state: 0, + handling_inputs_with: fn(ctx, state, input) { + case input { + Some(input) -> drift.continue(ctx, state + input) + None -> + ctx + |> drift.output(state) + |> drift.stop(state) + } + }, + ) + // Start the actor without wrapping the subject + |> actor.start(100, fn(subject) { subject }) + + let subject = started.data process.send(subject, Some(40)) process.send(subject, Some(2)) From d337781b5a7db4ad9740cbffcf3f182b9e19e5d5 Mon Sep 17 00:00:00 2001 From: Sakari Bergen Date: Sun, 20 Jul 2025 12:04:21 +0300 Subject: [PATCH 3/5] Bump version, add changelog, set tag prefix --- drift_actor/CHANGELOG.md | 20 ++++++++++++++++++++ drift_actor/README.md | 2 +- drift_actor/gleam.toml | 4 ++-- 3 files changed, 23 insertions(+), 3 deletions(-) create mode 100644 drift_actor/CHANGELOG.md diff --git a/drift_actor/CHANGELOG.md b/drift_actor/CHANGELOG.md new file mode 100644 index 0000000..c493e4a --- /dev/null +++ b/drift_actor/CHANGELOG.md @@ -0,0 +1,20 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## Unreleased + +### Changed +- Revamp the poorly designed API for starting actors. + Now supports named subjects, is safer, and should support future extensions better. + +## [1.0.0] - 2025-07-19 + +### Added +- Initial release with basic functionality. + +[Unreleased]: https://github.com/sbergen/spoke/compare/v1.0.0...HEAD +[1.0.0]: https://github.com/sbergen/spoke/releases/tag/v1.0.0 \ No newline at end of file diff --git a/drift_actor/README.md b/drift_actor/README.md index 0ee095f..1b163e0 100644 --- a/drift_actor/README.md +++ b/drift_actor/README.md @@ -8,7 +8,7 @@ ```sh gleam add gleam_erlang@1 gleam add drift@1 -gleam add drift_actor@1 +gleam add drift_actor@2 ``` ```gleam import drift diff --git a/drift_actor/gleam.toml b/drift_actor/gleam.toml index f58d91b..f5761a4 100644 --- a/drift_actor/gleam.toml +++ b/drift_actor/gleam.toml @@ -1,9 +1,9 @@ name = "drift_actor" -version = "1.0.0" +version = "2.0.0" description = "Run drift steppers as an OTP actor on Erlang." licences = ["Apache-2.0"] -repository = { type = "github", user = "sbergen", repo = "drift", path = "drift_actor" } +repository = { type = "github", user = "sbergen", repo = "drift", path = "drift_actor", tag-prefix = "drift_actor-" } [dependencies] gleam_stdlib = ">= 0.60.0 and < 2.0.0" From 540a9d0634619357eff88da38d5a50c08da64b5e Mon Sep 17 00:00:00 2001 From: Sakari Bergen Date: Sun, 20 Jul 2025 12:10:13 +0300 Subject: [PATCH 4/5] Make supervision test more stable --- drift_actor/test/drift/actor/supervision_test.gleam | 2 ++ 1 file changed, 2 insertions(+) diff --git a/drift_actor/test/drift/actor/supervision_test.gleam b/drift_actor/test/drift/actor/supervision_test.gleam index 3964336..26e4114 100644 --- a/drift_actor/test/drift/actor/supervision_test.gleam +++ b/drift_actor/test/drift/actor/supervision_test.gleam @@ -29,6 +29,8 @@ pub fn supervision_test() { // We should be able to use the same named subject again, // to kill the newly spawned child. process.send(inputs, Nil) + let assert Ok(_) = process.receive(children, 10) + as "Child should restart again" assert !process.is_alive(pid2) process.send_exit(supervisor.pid) From 314bf56092c76027745447d40637fbec1a9c0d9c Mon Sep 17 00:00:00 2001 From: Sakari Bergen Date: Sun, 20 Jul 2025 12:12:34 +0300 Subject: [PATCH 5/5] Update example --- .../src/catfacts_erlang/catfacts.gleam | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/examples/catfacts_erlang/src/catfacts_erlang/catfacts.gleam b/examples/catfacts_erlang/src/catfacts_erlang/catfacts.gleam index fdbc6c9..9140c3b 100644 --- a/examples/catfacts_erlang/src/catfacts_erlang/catfacts.gleam +++ b/examples/catfacts_erlang/src/catfacts_erlang/catfacts.gleam @@ -16,10 +16,15 @@ pub opaque type Catfacts { /// Starts a new cat facts actor. pub fn new() -> Catfacts { // The init really shouldn't fail, so we just assert success. - let assert Ok(actor) = - actor.using_io(new_io, fn(io_state) { io_state.selector }, handle_output) - |> actor.start(100, catfacts.new(), catfacts.handle_input) - Catfacts(actor) + let assert Ok(started) = + actor.using_io( + new_io, + selecting_inputs: fn(io_state) { io_state.selector }, + handling_outputs_with: handle_output, + ) + |> actor.with_stepper(catfacts.new(), catfacts.handle_input) + |> actor.start(100, Catfacts) + started.data } /// Wrapper function for performing the fetch in a blocking call,