In this series we are going to look at managing concurrency in Gleam in the following ways:

1. plain processes
2. actors and subjects and selectors(this post)
3. supervised actors (not yet published)

This series supersedes an earlier series of 3 posts that I made in 2024 on this same blog.

Actors

A gleam actor has no equivalent in Erlang. It is a process that receives statically typed messages. In erlang, a gen_server receives dynamically typed messages.

You would seldomly use an actor by itself in gleam. It is intened to keep receiving messages, in other words, suppposed to keep running. That means you will want it supervised. Because that is a large topic by itself, we are limiting ourselves to unsupervised actors in this post.

Subjects

Subjects are the statically typed channels over which the messages are sent to an actor. Responses return via their own subjects (they have to, responses to different messages may have different types).

Selectors

Selectors are a mechanism to combine messages from different types/subjects to be received by a single process.

Counter functionality

For all the examples that follow, we prepared a file containing counter functionality:

import gleam/erlang/process
import gleam/otp/actor

pub opaque type Msg {
  Up
  Down
  Get(response_channel: process.Subject(Int))
  Stop(response_channel: process.Subject(Nil))
}

pub fn up(counter: process.Subject(Msg)) {
  actor.send(counter, Up)
}

pub fn down(counter: process.Subject(Msg)) {
  actor.send(counter, Down)
}

pub fn get(counter: process.Subject(Msg)) -> Int {
  actor.call(counter, 123, Get)
}

pub fn stop(counter: process.Subject(Msg)) {
  actor.call(counter, 123, Stop)
}

pub fn loop(value: Int, msg: Msg) -> actor.Next(Int, Msg) {
  case msg {
    Up -> actor.continue(value + 1)

    Down -> actor.continue(value - 1)

    Get(response_channel) -> {
      process.send(response_channel, value)
      actor.continue(value)
    }

    Stop(response_channel) -> {
      process.send(response_channel, Nil)
      actor.stop()
    }
  }
}

Here you can see that the return subjects are part of the message sent to the actor. actor.call injects the subject during runtime.

When an actor continues its loop, of course it needs to provide the new state.

A simple actor

Let us start with a simple actor:

import counter
import gleam/otp/actor

pub fn start() {
  let assert Ok(actor.Started(_pid, counter)) =
    actor.new(0)
    |> actor.on_message(counter.loop)
    |> actor.start

  counter
}

The actor is created with the initial state, 0.

Whenever it receives a message, the function counter.loop is called.

Then we are ready to start the actor. This is, in this simple case, successful, and we obtain the erlang pid and counter, which is of type Subject(counter.Msg), that is, the subject over which we can send message to our actor.

An actor with a separate initialization

This actor performs exactly the same, we use this code to show how new_with_initialiser works.

import counter
import gleam/otp/actor

pub fn start() {
  let assert Ok(actor.Started(_pid, counter)) =
    actor.new_with_initialiser(123, fn(default_subject) {
      actor.initialised(0)
      |> actor.returning(default_subject)
      |> Ok
    })
    |> actor.on_message(counter.loop)
    |> actor.start

  counter
}

The first argument is a timeout in ms. The second argument is the initialiser function. The actor will fail to start when the initialiser function takes longer.

! This function is meant to set things up, and not run things that may fail or may be ran again later. You should not connect to a remote system that the actor will use; that actor should do that in the loop and reconnect when necessary. That way you can handle the disconnect (return error to caller; or retry). With a crash, your caller will also crash, and you will have an unhappy user.

The initial state of the actor is passed to the initialised function.

A default subject is passed to the initialiser function. If you want to use it, you must return it to the caller.

An actor with a named subject and a selector

Let us look ahead to supervised actors, that is, actors that crash and are restarted. Because only the process that created a subject can receive messages on it, the original subjects cannot be used to send messages to the new actor. The solution in gleam for this is a named subject. Here is how to set them up:

import counter
import gleam/erlang/process
import gleam/otp/actor

pub fn start(name: process.Name(counter.Msg)) {
  let assert Ok(actor.Started(_pid, counter)) =
    actor.new_with_initialiser(123, fn(_subject) {
      let subject = process.named_subject(name)
      let selector =
        process.new_selector()
        |> process.select(subject)

      actor.initialised(0)
      |> actor.selecting(selector)
      |> Ok
    })
    |> actor.named(name)
    |> actor.on_message(counter.loop)
    |> actor.start

  counter
}

The name is created with process.new_name("name") and is typed, as you can see. You must pass it around on setup, since another call will produce a different value.

This is an example where the default subject is not used; because a usable subject can be recreated from the name at will, we do not need to return it, either.

The actor module has no function to use a subject directly, so we have to use a selector, instead. A selector can receive from multiple subjects, with or without mapping (wrapping) the message. You could e.g. have a data subject and a control subject.

! The actor must be named with the same name that is used to create the subject(s) - this is how the runtime knows where to send the messages.

---

Although the construction of the subject differs, all three examples provide the same funtionality, as you can see in the program that runs them all:

import counter
import gleam/erlang/process
import initialized_actor
import selector_actor
import simple_actor

pub fn main() {
  simple_actor.start() |> run

  initialized_actor.start() |> run

  let n = process.new_name("stop_counter")
  let Nil = selector_actor.start(n)
  process.named_subject(n) |> run
}

fn run(c) {
  counter.up(c)
  assert counter.get(c) == 1
  assert counter.stop(c) == Nil
}