gleam/otp/actor

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.

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”.

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.

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:

// 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)
        }
      }
    }
  }
}

pub opaque type Builder(state, message, return)

pub opaque type Initialised(state, message, data)

pub opaque type Next(state, message)

pub type StartError {
  InitTimeout
  InitFailed(String)
  InitExited(process.ExitReason)
}

pub type StartResult(data) =
  Result(Started(data), StartError)

pub type Started(data) {
  Started(pid: process.Pid, data: data)
}

pub fn call(
  subject: process.Subject(message),
  timeout: Int,
  make_message: fn(process.Subject(reply)) -> message,
) -> reply

pub fn continue(state: state) -> Next(state, message)

pub fn initialised(
  state: state,
) -> Initialised(state, message, Nil)

pub fn named(
  builder: Builder(state, message, return),
  name: process.Name(message),
) -> Builder(state, message, return)

pub fn new(
  state: state,
) -> Builder(state, message, process.Subject(message))

pub fn new_with_initialiser(
  timeout: Int,
  initialise: fn(process.Subject(message)) -> Result(
    Initialised(state, message, return),
    String,
  ),
) -> Builder(state, message, return)

pub fn on_message(
  builder: Builder(state, message, return),
  handler: fn(state, message) -> Next(state, message),
) -> Builder(state, message, return)

pub fn returning(
  initialised: Initialised(state, message, old_return),
  return: return,
) -> Initialised(state, message, return)

pub fn selecting(
  initialised: Initialised(state, old_message, return),
  selector: process.Selector(message),
) -> Initialised(state, message, return)

pub fn send(subject: process.Subject(msg), msg: msg) -> Nil

pub fn start(
  builder: Builder(state, msg, return),
) -> Result(Started(return), StartError)

pub fn stop() -> Next(state, message)

pub fn stop_abnormal(reason: String) -> Next(state, message)

pub fn with_selector(
  value: Next(state, message),
  selector: process.Selector(message),
) -> Next(state, message)