Groups

Groups are named sets of topics on the server. Use one group broadcast to send an event to many topics. You do not need to track subscriptions. Groups are similar to Socket.IO rooms and SignalR groups.

Start the groups actor

import beryl/group
import gleam/otp/static_supervisor

let #(groups, groups_specification) = group.child_spec()
let assert Ok(_root) =
  static_supervisor.new(static_supervisor.OneForOne)
  |> static_supervisor.add(groups_specification)
  |> static_supervisor.start()

group.child_spec() returns a stable Groups handle. Add its child specification to the application supervisor before you use the handle.

Synchronous group operations wait up to 5 seconds for the actor by default. Configure a different timeout when starting the actor:

let config =
  group.default_config()
  |> group.with_call_timeout(10_000)
let #(groups, groups_specification) = group.child_spec_with_config(config)
let assert Ok(_root) =
  static_supervisor.new(static_supervisor.OneForOne)
  |> static_supervisor.add(groups_specification)
  |> static_supervisor.start()

create, delete, add, remove, topics, and list_groups return Error(GroupCallFailed(reason)) if the actor is unavailable, exits during the call, or does not reply within this timeout. A timed-out mutation may still take effect. broadcast first performs the same synchronous topic lookup, then sends each topic broadcast without waiting for delivery.

Creating and deleting groups

// Create a group
let assert Ok(Nil) = group.create(groups, "team:engineering")

// The error includes the name that is already in use.
case group.create(groups, "team:engineering") {
  Ok(Nil) -> Nil
  Error(group.GroupAlreadyExists(name)) -> Nil
  Error(group.GroupNotFound(_)) -> Nil  // cannot occur for create
  Error(group.GroupCallFailed(reason)) -> handle_call_failure(reason)
}

// Delete a group (removes it and all its topic memberships)
let assert Ok(Nil) = group.delete(groups, "team:engineering")

// The error includes the name that was not found.
case group.delete(groups, "team:gone") {
  Ok(Nil) -> Nil
  Error(group.GroupNotFound(name)) -> Nil
  Error(group.GroupAlreadyExists(_)) -> Nil  // cannot occur for delete
  Error(group.GroupCallFailed(reason)) -> handle_call_failure(reason)
}

Adding and removing topics

Topics are strings that match channel topics. Groups do not check whether a topic has subscribers. They store sets of strings.

let assert Ok(Nil) = group.add(groups, "team:engineering", "room:frontend")
let assert Ok(Nil) = group.add(groups, "team:engineering", "room:backend")
let assert Ok(Nil) = group.add(groups, "team:engineering", "room:infra")

// Remove one topic
let assert Ok(Nil) = group.remove(groups, "team:engineering", "room:infra")

// Both add and remove return Error(GroupNotFound) if the group doesn't exist

Adding the same topic twice does nothing because the group stores a set.

Read groups

// List all topics in a group
case group.topics(groups, "team:engineering") {
  Ok(topic_set) -> set.to_list(topic_set)  // ["room:frontend", "room:backend"]
  Error(group.GroupNotFound(_)) -> []
  Error(group.GroupAlreadyExists(_)) -> []  // cannot occur for topics
  Error(group.GroupCallFailed(reason)) -> handle_lookup_failure(reason)
}

// List all group names
let assert Ok(names) = group.list_groups(groups)  // ["team:engineering", "team:design"]

Send an event to every topic in a group

group.broadcast asks the groups actor for the topic set. The calling process waits for that lookup, then sends the event to each topic with beryl.broadcast. The actor does not send the broadcasts itself. The function returns an admission Result. A missing group returns GroupLookupFailed(GroupNotFound(name)).

let assert Ok(Nil) = group.broadcast(
  groups,
  channels,
  "team:engineering",
  "deploy_started",
  json.object([#("env", json.string("production"))]),
)

This has the same effect as one beryl.broadcast call for each group topic.

Group errors

Error	When
`GroupAlreadyExists(name)`	`create` called for a name already in use
`GroupNotFound(name)`	`delete`, `add`, `remove`, or `topics` called for an unknown group name
`GroupCallFailed(reason)`	The actor is unavailable, exits during a call, or exceeds the call timeout

beryl/error.CallError distinguishes CallUnavailable, CallDown(reason), and CallTimedOut. The actor-down error retains its exit reason.

Complete example: team rooms

import beryl
import beryl/group
import gleam/json
import gleam/otp/static_supervisor

// At startup
let #(groups, groups_specification) = group.child_spec()
let assert Ok(_root) =
  static_supervisor.new(static_supervisor.OneForOne)
  |> static_supervisor.add(groups_specification)
  |> static_supervisor.start()
let assert Ok(Nil) = group.create(groups, "team:eng")
let assert Ok(Nil) = group.add(groups, "team:eng", "room:frontend")
let assert Ok(Nil) = group.add(groups, "team:eng", "room:backend")

// Later: broadcast deployment notice to all engineering rooms
let assert Ok(Nil) = group.broadcast(
  groups,
  channels,
  "team:eng",
  "deploy_complete",
  json.object([
    #("version", json.string("1.4.2")),
    #("deployed_by", json.string("ci")),
  ]),
)

// When a team is disbanded
let assert Ok(Nil) = group.delete(groups, "team:eng")

Restarts and node limits

Start the groups actor with group.child_spec. Its handle uses a stable registered name and reaches the replacement actor after a supervised restart. The actor keeps group definitions and topic memberships in memory. A restart clears them.

The registered name is node-local. Keep a Groups handle on the node where its child specification runs. From another BEAM node, synchronous operations cannot reach the owning actor and return a call error. broadcast wraps that error in GroupLookupFailed. Group definitions and memberships are not replicated between nodes.

See the Supervision guide for the overall startup pattern.