Gleam Actor Model

What this skill does

# Gleam Actor Model

## Introduction

Gleam leverages the Erlang VM's actor model, enabling lightweight concurrent
processes that communicate through message passing. This model provides inherent
fault tolerance, isolation, and scalability, making it ideal for building
distributed systems.

The actor model in Gleam uses OTP (Open Telecom Platform) patterns including
GenServers for stateful processes, supervisors for fault recovery, and message
passing for inter-process communication. Each process has its own heap and
communicates asynchronously, eliminating shared memory concerns.

This skill covers process creation and message passing, GenServer pattern for
stateful actors, supervisors and fault tolerance, process linking and monitoring,
selective receive, and patterns for building robust concurrent applications.

## Process Basics and Message Passing

Processes are lightweight, isolated units of execution that communicate via
message passing.

```gleam
import gleam/erlang/process
import gleam/io

// Basic process creation
pub fn simple_process() {
  process.spawn(fn() {
    io.println("Hello from process!")
  })
}

// Process with message passing
pub type Message {
  Ping
  Pong
  Stop
}

pub fn echo_process() {
  let subject = process.new_subject()

  process.spawn(fn() {
    loop(subject)
  })

  subject
}

fn loop(subject: process.Subject(Message)) {
  case process.receive(subject, 1000) {
    Ok(Ping) -> {
      io.println("Received Ping")
      loop(subject)
    }
    Ok(Pong) -> {
      io.println("Received Pong")
      loop(subject)
    }
    Ok(Stop) -> {
      io.println("Stopping")
      Nil
    }
    Error(_) -> {
      io.println("Timeout")
      loop(subject)
    }
  }
}

// Sending messages
pub fn send_messages(subject: process.Subject(Message)) {
  process.send(subject, Ping)
  process.send(subject, Pong)
  process.send(subject, Stop)
}

// Request-response pattern
pub type Request {
  GetValue(reply_to: process.Subject(Int))
  SetValue(value: Int, reply_to: process.Subject(Nil))
}

pub fn state_process(initial: Int) {
  let subject = process.new_subject()

  process.spawn(fn() {
    state_loop(subject, initial)
  })

  subject
}

fn state_loop(subject: process.Subject(Request), state: Int) {
  case process.receive(subject, 5000) {
    Ok(GetValue(reply_to)) -> {
      process.send(reply_to, state)
      state_loop(subject, state)
    }
    Ok(SetValue(value, reply_to)) -> {
      process.send(reply_to, Nil)
      state_loop(subject, value)
    }
    Error(_) -> state_loop(subject, state)
  }
}

// Calling the state process
pub fn use_state_process() {
  let proc = state_process(0)
  let reply_subject = process.new_subject()

  // Set value
  process.send(proc, SetValue(42, reply_subject))
  let _ack = process.receive(reply_subject, 1000)

  // Get value
  process.send(proc, GetValue(reply_subject))
  case process.receive(reply_subject, 1000) {
    Ok(value) -> io.debug(value)
    Error(_) -> io.println("Timeout")
  }
}

// Process with multiple message types
pub type ServerMessage {
  Request(id: Int, reply_to: process.Subject(String))
  Broadcast(message: String)
  Shutdown
}

pub fn multi_message_process() {
  let subject = process.new_subject()

  process.spawn(fn() {
    multi_loop(subject, [])
  })

  subject
}

fn multi_loop(
  subject: process.Subject(ServerMessage),
  clients: List(process.Subject(String)),
) {
  case process.receive(subject, 1000) {
    Ok(Request(id, reply_to)) -> {
      let response = "Response for " <> int.to_string(id)
      process.send(reply_to, response)
      multi_loop(subject, [reply_to, ..clients])
    }
    Ok(Broadcast(message)) -> {
      list.each(clients, fn(client) {
        process.send(client, message)
      })
      multi_loop(subject, clients)
    }
    Ok(Shutdown) -> Nil
    Error(_) -> multi_loop(subject, clients)
  }
}

// Process pools
pub fn worker_pool(size: Int) -> List(process.Subject(Message)) {
  list.range(1, size)
  |> list.map(fn(_) { echo_process() })
}

pub fn distribute_work(pool: List(process.Subject(Message)),
                       work: List(Message)) {
  list.zip(work, list.cycle(pool))
  |> list.each(fn(pair) {
    let #(message, worker) = pair
    process.send(worker, message)
  })
}
```

Lightweight processes with message passing enable concurrent applications without
shared memory complexity.

## GenServer Pattern

GenServer provides a standard pattern for stateful processes with synchronous
and asynchronous operations.

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

// State type
pub type Counter {
  Counter(value: Int)
}

// Message types
pub type CounterMessage {
  Increment
  Decrement
  GetValue(reply_to: process.Subject(Int))
  Reset(reply_to: process.Subject(Nil))
}

// GenServer implementation
pub fn start_counter() -> Result(process.Subject(CounterMessage),
                                 actor.StartError) {
  actor.start(Counter(value: 0), handle_message)
}

fn handle_message(
  message: CounterMessage,
  state: Counter,
) -> actor.Next(CounterMessage, Counter) {
  case message {
    Increment -> {
      actor.continue(Counter(value: state.value + 1))
    }
    Decrement -> {
      actor.continue(Counter(value: state.value - 1))
    }
    GetValue(reply_to) -> {
      process.send(reply_to, state.value)
      actor.continue(state)
    }
    Reset(reply_to) -> {
      process.send(reply_to, Nil)
      actor.continue(Counter(value: 0))
    }
  }
}

// Using the GenServer
pub fn use_counter() {
  case start_counter() {
    Ok(counter) -> {
      // Increment
      process.send(counter, Increment)
      process.send(counter, Increment)

      // Get value
      let reply = process.new_subject()
      process.send(counter, GetValue(reply))
      case process.receive(reply, 1000) {
        Ok(value) -> io.debug(value)  // 2
        Error(_) -> io.println("Timeout")
      }
    }
    Error(_) -> io.println("Failed to start counter")
  }
}

// GenServer with complex state
pub type CacheState {
  CacheState(items: Dict(String, String), max_size: Int)
}

pub type CacheMessage {
  Get(key: String, reply_to: process.Subject(Option(String)))
  Put(key: String, value: String, reply_to: process.Subject(Bool))
  Delete(key: String, reply_to: process.Subject(Bool))
  Size(reply_to: process.Subject(Int))
}

pub fn start_cache(max_size: Int) -> Result(process.Subject(CacheMessage),
                                            actor.StartError) {
  actor.start(
    CacheState(items: dict.new(), max_size: max_size),
    handle_cache_message,
  )
}

fn handle_cache_message(
  message: CacheMessage,
  state: CacheState,
) -> actor.Next(CacheMessage, CacheState) {
  case message {
    Get(key, reply_to) -> {
      let value = dict.get(state.items, key)
      process.send(reply_to, value)
      actor.continue(state)
    }
    Put(key, value, reply_to) -> {
      let current_size = dict.size(state.items)
      case current_size < state.max_size {
        True -> {
          let new_items = dict.insert(state.items, key, value)
          process.send(reply_to, True)
          actor.continue(CacheState(..state, items: new_items))
        }
        False -> {
          process.send(reply_to, False)
          actor.continue(state)
        }
      }
    }
    Delete(key, reply_to) -> {
      let new_items = dict.delete(state.items, key)
      process.send(reply_to, True)
      actor.continue(CacheState(..state, items: new_items))
    }
    Size(reply_to) -> {
      process.send(reply_to, dict.size(state.items))
      actor.continue(state)
    }
  }
}

// GenServer with initialization
pub type ConnectionState {
  ConnectionState(url: String, connected: Bool)
}

pub type ConnectionMessage {
  Connect(reply_to: process.Subject(Result(Nil, String)))
  Disconnect
  Status(reply_to: process.Subject(Bool))
}

pub fn start_connection(url: String) ->
  Result(process.Subject(ConnectionMessage), actor.StartError) {
  actor.start_spec(actor.Spec(
    init: fn() {
      // Ini