//// This module provides the _Actor_ abstraction, one of the most common //// building blocks of Gleam OTP programs. //// //// An Actor is a process like any other BEAM process and can be used to hold //// state, execute code, and communicate with other processes by sending and //// receiving messages. The advantage of using the actor abstraction over a bare //// process is that it provides a single interface for commonly needed //// functionality, including support for the [tracing and debugging //// features in OTP][erlang-sys]. //// //// Gleam's Actor is similar to Erlang's `gen_server` and Elixir's `GenServer` //// but differs in that it offers a fully typed interface. This different API is //// why Gleam uses the name "Actor" rather than some variation of //// "generic-server". //// //// [erlang-sys]: https://www.erlang.org/doc/man/sys.html //// //// ## Example //// //// An Actor can be used to create a client-server interaction between an Actor //// (the server) and other processes (the clients). In this example we have an //// Actor that works as a stack, allowing clients to push and pop elements. //// //// ```gleam //// pub fn main() { //// // Start the actor with initial state of an empty list, and the //// // `handle_message` callback function (defined below). //// // We assert that it starts successfully. //// // //// // In real-world Gleam OTP programs we would likely write a wrapper functions //// // called `start`, `push` `pop`, `shutdown` to start and interact with the //// // Actor. We are not doing that here for the sake of showing how the Actor //// // API works. //// let assert Ok(actor) = //// actor.new([]) |> actor.on_message(handle_message) |> actor.start //// let subject = actor.data //// //// // We can send a message to the actor to push elements onto the stack. //// process.send(subject, Push("Joe")) //// process.send(subject, Push("Mike")) //// process.send(subject, Push("Robert")) //// //// // The `Push` message expects no response, these messages are sent purely for //// // the side effect of mutating the state held by the actor. //// // //// // We can also send the `Pop` message to take a value off of the actor's //// // stack. This message expects a response, so we use `process.call` to send a //// // message and wait until a reply is received. //// // //// // In this instance we are giving the actor 10 milliseconds to reply, if the //// // `call` function doesn't get a reply within this time it will panic and //// // crash the client process. //// let assert Ok("Robert") = process.call(subject, 10, Pop) //// let assert Ok("Mike") = process.call(subject, 10, Pop) //// let assert Ok("Joe") = process.call(subject, 10, Pop) //// //// // The stack is now empty, so if we pop again the actor replies with an error. //// let assert Error(Nil) = process.call(subject, 10, Pop) //// //// // Lastly, we can send a message to the actor asking it to shut down. //// process.send(subject, Shutdown) //// } //// ``` //// //// Here is the code that is used to implement this actor: //// //// ```gleam //// // First step of implementing the stack Actor is to define the message type that //// // it can receive. //// // //// // The type of the elements in the stack is not fixed so a type parameter //// // is used for it instead of a concrete type such as `String` or `Int`. //// pub type Message(element) { //// // The `Shutdown` message is used to tell the actor to stop. //// // It is the simplest message type, it contains no data. //// // //// // Most the time we don't define an API to shut down an actor, but in this //// // example we do to show how it can be done. //// Shutdown //// //// // The `Push` message is used to add a new element to the stack. //// // It contains the item to add, the type of which is the `element` //// // parameterised type. //// Push(push: element) //// //// // The `Pop` message is used to remove an element from the stack. //// // It contains a `Subject`, which is used to send the response back to the //// // message sender. In this case the reply is of type `Result(element, Nil)`. //// Pop(reply_with: Subject(Result(element, Nil))) //// } //// //// // The last part is to implement the `handle_message` callback function. //// // //// // This function is called by the Actor each for each message it receives. //// // Actors are single threaded only does one thing at a time, so they handle //// // messages sequentially and one at a time, in the order they are received. //// // //// // The function takes the message and the current state, and returns a data //// // structure that indicates what to do next, along with the new state. //// fn handle_message( //// stack: List(e), //// message: Message(e), //// ) -> actor.Next(List(e), Message(e)) { //// case message { //// // For the `Shutdown` message we return the `actor.stop` value, which causes //// // the actor to discard any remaining messages and stop. //// // We may chose to do some clean-up work here, but this actor doesn't need //// // to do this. //// Shutdown -> actor.stop() //// //// // For the `Push` message we add the new element to the stack and return //// // `actor.continue` with this new stack, causing the actor to process any //// // queued messages or wait for more. //// Push(value) -> { //// let new_state = [value, ..stack] //// actor.continue(new_state) //// } //// //// // For the `Pop` message we attempt to remove an element from the stack, //// // sending it or an error back to the caller, before continuing. //// Pop(client) -> { //// case stack { //// [] -> { //// // When the stack is empty we can't pop an element, so we send an //// // error back. //// process.send(client, Error(Nil)) //// actor.continue([]) //// } //// //// [first, ..rest] -> { //// // Otherwise we send the first element back and use the remaining //// // elements as the new state. //// process.send(client, Ok(first)) //// actor.continue(rest) //// } //// } //// } //// } //// } //// ``` import gleam/dynamic.{type Dynamic} import gleam/erlang/atom import gleam/erlang/charlist.{type Charlist} import gleam/erlang/process.{ type ExitReason, type Pid, type Selector, type Subject, Abnormal, Killed, } import gleam/option.{type Option, None, Some} import gleam/otp/system.{ type DebugState, type Mode, type StatusInfo, type SystemMessage, GetState, GetStatus, Resume, Running, StatusInfo, Suspend, Suspended, } import gleam/result import gleam/string type Message(message) { /// A regular message excepted by the process Message(message) /// An OTP system message, for debugging or maintenance System(SystemMessage) /// An unexpected message Unexpected(Dynamic) } /// The type used to indicate what to do after handling a message. /// pub opaque type Next(state, message) { /// Continue handling messages. /// /// An optional selector can be provided to changes the messages that the /// actor is handling. This replaces any selector that was previously given /// in the actor's `init` callback, or in any previous `Next` value. /// Continue(state: state, selector: Option(Selector(message))) /// Stop handling messages and shut down. /// Stop(ExitReason) } /// Indicate the actor should continue, processing any waiting or future messages. /// pub fn continue(state: state) -> Next(state, message) { Continue(state:, selector: None) } /// Indicate the actor should stop and shut-down, handling no futher messages. /// /// The reason for exiting is `Normal`. /// pub fn stop() -> Next(state, message) { Stop(process.Normal) } /// Provide a selector to change the messages that the actor is handling /// going forward. This replaces any selector that was previously given /// in the actor's `init` callback, or in any previous `Next` value. /// pub fn with_selector( value: Next(state, message), selector: Selector(message), ) -> Next(state, message) { case value { Continue(state, _) -> Continue(state, Some(selector)) Stop(_) -> value } } type Self(state, msg) { Self( /// The mode the actor is currently in, either active or suspended. mode: Mode, /// The pid of the process that started this actor. parent: Pid, /// The state of this actor, provided by the programmer. state: state, /// The selector that actor is currently using to reveive messages. This /// can be changed by the `Next` value returned by the actor's `loop` callback. selector: Selector(Message(msg)), /// An opaque value used by the OTP system debug APIs. debug_state: DebugState, /// The message handling code provided by the programmer. message_handler: fn(state, msg) -> Next(state, msg), ) } /// A value returned to the parent when their child actor successfully starts. pub type Started(data) { Started( /// The process identifier of the started actor. This can be used to /// monitor the actor, make it exit, or anything else you can do with a /// pid. pid: Pid, /// Data returned by the actor after it initialised. Commonly this will be /// a subject that it will receive messages from. data: data, ) } /// A convenience for the type returned when an actor process is started. pub type StartResult(data) = Result(Started(data), StartError) /// A type returned from an actor's initialiser, containing the actor state, a /// selector to receive messages using, and data to return to the parent. /// /// Use the `initialised`, `selecting`, and `returning` functions to construct /// this type. /// pub opaque type Initialised(state, message, data) { Initialised(state: state, selector: Option(Selector(message)), return: data) } /// Takes the post-initialisation state of the actor. This state will be passed /// to the `on_message` callback each time a message is received. /// pub fn initialised(state: state) -> Initialised(state, message, Nil) { Initialised(state, None, Nil) } /// Add a selector for the actor to receive messages with. /// /// If a message is received by the actor but not selected for with the /// selector then the actor will discard it and log a warning. /// pub fn selecting( initialised: Initialised(state, old_message, return), selector: Selector(message), ) -> Initialised(state, message, return) { Initialised(..initialised, selector: Some(selector)) } /// Add the data to return to the parent process. This might be a subject that /// the actor will receive messages over. /// pub fn returning( initialised: Initialised(state, message, old_return), return: return, ) -> Initialised(state, message, return) { Initialised(..initialised, return:) } pub opaque type Builder(state, message, return) { Builder( /// The initialisation functionality for the actor. This function is called /// just after the actor starts but before the channel sender is returned /// to the parent. /// /// This function is used to ensure that any required data or state is /// correct. If this function returns an error it means that the actor has /// failed to start and an error is returned to the parent. /// initialise: fn(Subject(message)) -> Result(Initialised(state, message, return), String), /// How many milliseconds the `init` function has to return before it is /// considered to have taken too long and failed. /// initialisation_timeout: Int, /// This function is called to handle each message that the actor receives. /// on_message: fn(state, message) -> Next(state, message), /// The actor can be named for you at start. /// name: Option(process.Name(message)), ) } /// Create a builder for an actor without a custom initialiser. The actor /// returns a subject to the parent that can be used to send messages to the /// actor. /// /// If the actor has been given a name with the `named` function then the /// subject is a named subject. /// /// If you wish to create an actor with some other initialisation logic that /// runs before it starts handling messages, see `new_with_initialiser`. /// pub fn new(state: state) -> Builder(state, message, Subject(message)) { let initialise = fn(subject) { initialised(state) |> returning(subject) |> Ok } Builder( initialise:, initialisation_timeout: 1000, on_message: fn(state, _) { continue(state) }, name: option.None, ) } /// Create a builder for an actor with a custom initialiser that runs before /// the start function returns to the parent, and before the actor starts /// handling messages. /// /// The first argument is a number of milliseconds that the initialiser /// function is expected to return within. If it takes longer the initialiser /// is considered to have failed and the actor will be killed, and an error /// will be returned to the parent. /// /// The actor's default subject is passed to the initialiser function. You can /// chose to return it to the parent with `returning`, use it in some other /// way, or ignore it completely. /// /// If a custom selector is given using the `selecting` function then this /// overwrites the default selector, which selects for the default subject, so /// you will need to add the subject to the custom selector yourself. /// pub fn new_with_initialiser( timeout: Int, initialise: fn(Subject(message)) -> Result(Initialised(state, message, return), String), ) -> Builder(state, message, return) { Builder( initialise:, initialisation_timeout: timeout, on_message: fn(state, _) { continue(state) }, name: option.None, ) } /// Set the message handler for the actor. This callback function will be /// called each time the actor receives a message. /// /// Actors handle messages sequentially, later messages being handled after the /// previous one has been handled. pub fn on_message( builder: Builder(state, message, return), handler: fn(state, message) -> Next(state, message), ) -> Builder(state, message, return) { Builder(..builder, on_message: handler) } /// Provide a name for the actor to be registered with when started, enabling /// it to receive messages via a named subject. This is useful for making /// processes that can take over from an older one that has exited due to a /// failure, or to avoid passing subjects from receiver processes to sender /// processes. /// /// If the name is already registered to another process then the actor will /// fail to start. /// /// When this function is used the actor's default subject will be a named /// subject using this name. /// pub fn named( builder: Builder(state, message, return), name: process.Name(message), ) -> Builder(state, message, return) { Builder(..builder, name: option.Some(name)) } fn exit_process(reason: ExitReason) -> ExitReason { case reason { Abnormal(reason) -> process.send_abnormal_exit(process.self(), reason) Killed -> process.kill(process.self()) _ -> Nil } reason } fn receive_message(self: Self(state, msg)) -> Message(msg) { let selector = case self.mode { // When suspended we only respond to system messages Suspended -> process.new_selector() |> select_system_messages // When running we respond to all messages Running -> // The actor needs to handle various different messages: // // - OTP system messages. These are handled by the actor for the // programmer, they don't need to do anything. // - Messages sent to the subject the actor creates during initialisation // and returns to the parent. // - Any arbitrary messages the programmer expects the actor to receive. // For example, messages sent by a pubsub system where it does not // support using the actor's subject. // - Any unexpected messages. // // We add the handler for unexpected messages first so that the user // supplied selector can override it if desired. process.new_selector() |> process.select_other(Unexpected) |> process.merge_selector(self.selector) |> select_system_messages } process.selector_receive_forever(selector) } fn select_system_messages( selector: Selector(Message(msg)), ) -> Selector(Message(msg)) { selector |> process.select_record(atom.create("system"), 2, convert_system_message) } @external(erlang, "gleam_otp_external", "convert_system_message") fn convert_system_message(b: Dynamic) -> Message(msg) fn process_status_info(self: Self(state, msg)) -> StatusInfo { StatusInfo( module: atom.create("gleam@otp@actor"), parent: self.parent, mode: self.mode, debug_state: self.debug_state, state: dynamic.from(self.state), ) } fn loop(self: Self(state, msg)) -> ExitReason { case receive_message(self) { // An OTP system message. This is handled by the actor for the programmer, // behind the scenes. System(system) -> case system { GetState(callback) -> { callback(dynamic.from(self.state)) loop(self) } Resume(callback) -> { callback() loop(Self(..self, mode: Running)) } Suspend(callback) -> { callback() loop(Self(..self, mode: Suspended)) } GetStatus(callback) -> { callback(process_status_info(self)) loop(self) } } // An unexpected message. It this is reached then the programmer has not // handled this, so log a warning. Unexpected(message) -> { log_warning( charlist.from_string("Actor discarding unexpected message: ~s"), [charlist.from_string(string.inspect(message))], ) loop(self) } // A regular message that the programmer is expecting, either over the // subject or some other messsage that the programmer's selector expects. Message(msg) -> case self.message_handler(self.state, msg) { Stop(reason) -> exit_process(reason) Continue(state: state, selector: new_selector) -> { let selector = case new_selector { None -> self.selector Some(s) -> process.map_selector(s, Message) } loop(Self(..self, state: state, selector: selector)) } } } } // TODO: replace this when we have Gleam bindings to the logger @external(erlang, "logger", "warning") fn log_warning(a: Charlist, b: List(Charlist)) -> Nil // Run automatically when the actor is first started. fn initialise_actor( builder: Builder(state, msg, return), parent: Pid, ack: Subject(Result(return, String)), ) -> ExitReason { // Run the actor initialiser. let result = { use subject <- result.try(case builder.name { None -> Ok(process.new_subject()) Some(name) -> { use _ <- result.try(try_register_self(name)) Ok(process.named_subject(name)) } }) use result <- result.try(builder.initialise(subject)) Ok(#(subject, result)) } case result { // Init was OK, send the subject to the parent and start handling messages. Ok(#(subject, Initialised(state:, selector:, return:))) -> { // Add the default subject to the selector provided by the initialiser. // The initialiser may have added additional handlers to the selector. let selector = case selector { Some(selector) -> selector None -> process.new_selector() |> process.select(subject) } let selector = process.map_selector(selector, Message) // Signal to parent that the process has initialised successfully process.send(ack, Ok(return)) // Start message receive loop let self = Self( state: state, parent:, selector: selector, message_handler: builder.on_message, debug_state: system.debug_state([]), mode: Running, ) loop(self) } // The init failed. Send the reason back to the parent, but exit normally. Error(reason) -> { process.send(ack, Error(reason)) exit_process(process.Normal) } } } fn try_register_self(name: process.Name(msg)) -> Result(Nil, String) { case process.register(process.self(), name) { Ok(Nil) -> Ok(Nil) Error(_) -> Error("name already registered") } } pub type StartError { InitTimeout InitFailed(String) InitExited(ExitReason) } type StartInitMessage(data) { Ack(Result(data, String)) Mon(process.Down) } /// Start an actor from a given specification. If the actor's `init` function /// returns an error or does not return within `init_timeout` then an error is /// returned. /// /// If you do not need to specify the initialisation behaviour of your actor /// consider using the `start` function. /// pub fn start( builder: Builder(state, msg, return), ) -> Result(Started(return), StartError) { let timeout = builder.initialisation_timeout let ack_subject = process.new_subject() let self = process.self() let child = process.spawn(fn() { initialise_actor(builder, self, ack_subject) }) let monitor = process.monitor(child) let selector = process.new_selector() |> process.select_map(ack_subject, Ack) |> process.select_specific_monitor(monitor, Mon) let result = case process.selector_receive(selector, timeout) { // Child started OK Ok(Ack(Ok(subject))) -> Ok(subject) // Child initialiser returned an error Ok(Ack(Error(reason))) -> Error(InitFailed(reason)) // Child went down while initialising Ok(Mon(down)) -> Error(InitExited(down.reason)) // Child did not finish initialising in time Error(Nil) -> { // Unlink the child before killing it, so that we only return the error, // but don't also send an exit message to the linked parent process. process.unlink(child) process.kill(child) Error(InitTimeout) } } // Remove the monitor used for the starting of the actor as to avoid an extra // message arriving at the parent if the child dies later. process.demonitor_process(monitor) case result { Ok(data) -> Ok(Started(pid: child, data:)) Error(error) -> Error(error) } } /// Send a message over a given channel. /// /// This is a re-export of `process.send`, for the sake of convenience. /// pub fn send(subject: Subject(msg), msg: msg) -> Nil { process.send(subject, msg) } /// Send a synchronous message and wait for a response from the receiving /// process. /// /// If a reply is not received within the given timeout then the sender process /// crashes. If you wish to receive a `Result` rather than crashing see the /// `process.try_call` function. /// /// This is a re-export of `process.call`, for the sake of convenience. /// pub fn call( subject: Subject(message), timeout: Int, make_message: fn(Subject(reply)) -> message, ) -> reply { process.call(subject, timeout, make_message) }