Revamp actor starting

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" }

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

drift_actor/README.md

@@ -8,39 +8,47 @@
 ```sh
 gleam add gleam_erlang@1
 gleam add drift@1
-gleam add drift_actor@1
+gleam add drift_actor@2
 ```
 ```gleam
 import drift
 import drift/actor
 import gleam/erlang/process
-import gleam/function
 import gleam/io
 import gleam/option.{None, Some}
 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))

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"

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,

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

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
 }

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,22 @@ 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)
+  let assert Ok(_) = process.receive(children, 10)
+    as "Child should restart again"
+  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 +47,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
+    })
   })
 }