beryl/group

Topic groups

Groups organize topics for broadcasts to several topics at once, such as every channel in a team or a system-wide notification.

Groups are independent of the beryl runtime and run under your application's supervision tree.

Example

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:engineering")
let assert Ok(Nil) = group.add(groups, "team:engineering", "room:frontend")
let assert Ok(Nil) = group.add(groups, "team:engineering", "room:backend")
group.broadcast(groups, channels, "team:engineering", "announce", payload)

Types

• BroadcastError
• Config
• GroupError
• Groups
• Message

Functions

• add
• broadcast
• child_spec
• child_spec_with_config
• create
• default_config
• delete
• list_groups
• remove
• topics
• with_call_timeout

Types

BroadcastError

pub type BroadcastError {
  GroupLookupFailed(GroupError)
  BroadcastRejected(
    admitted_topics: Int,
    reason: overload.AdmissionError
  )
}

A group lookup or one topic's local broadcast admission failed.

Constructors

BroadcastRejected

BroadcastRejected(
  admitted_topics: Int,
  reason: overload.AdmissionError
)

Earlier topics were admitted; the failing topic and later topics were not.

Config

pub type Config

Configuration for starting a groups actor.

Build configs with default_config and the with_* functions.

GroupError

pub type GroupError {
  GroupAlreadyExists(name: String)
  GroupNotFound(name: String)
  GroupCallFailed(error.CallError)
}

Errors from group operations.

Constructors

GroupAlreadyExists

GroupAlreadyExists(name: String)

The group already exists.

GroupNotFound

GroupNotFound(name: String)

The group was not found.

GroupCallFailed

GroupCallFailed(error.CallError)

The actor was unavailable, exited, or did not reply in time.

Groups

pub type Groups

A running groups instance.

This handle is opaque. Callers cannot forge the actor subject or depend on its runtime representation.

Node affinity

The stable registered subject is resolved on the caller's node. Keep a Groups handle on the node where its child specification runs. Calls from another BEAM node cannot reach the owning actor. All public operations, including broadcast, return a typed call failure if the actor is unavailable or the call times out.

Message

pub type Message

Messages that the groups actor handles.

Functions

add

pub fn add(
  Groups,
  String,
  String
) -> Result(Nil, GroupError)

Add a topic to a group.

Returns GroupCallFailed if the actor is unavailable or does not reply within the configured call timeout (5 seconds by default).

broadcast

pub fn broadcast(
  Groups,
  beryl.Sockets,
  String,
  String,
  json.Json
) -> Result(Nil, BroadcastError)

Broadcast a message to all topics in a group.

This function sends the message to each topic through beryl.broadcast. The groups actor performs the topic lookup. The caller performs the fan-out. A missing group returns GroupLookupFailed. On admission failure, BroadcastRejected reports the number of earlier topics admitted. Those admissions remain valid; fan-out is not transactional.

Actor-call failures are returned inside GroupLookupFailed.

child_spec

pub fn child_spec() -> #(Groups, supervision.ChildSpecification(process.Subject(Message)))

Build the supervised groups actor with the default configuration.

Add the returned child specification to your application's supervisor. The returned handle is name-backed, so it reaches the replacement actor after a supervised restart. Group definitions are in-memory state and are reset by a restart.

child_spec_with_config

pub fn child_spec_with_config(Config) -> #(Groups, supervision.ChildSpecification(process.Subject(Message)))

Build the supervised groups actor with a custom configuration.

create

pub fn create(
  Groups,
  String
) -> Result(Nil, GroupError)

Create a named group.

Returns GroupCallFailed if the actor is unavailable or does not reply within the configured call timeout (5 seconds by default).

default_config

pub fn default_config() -> Config

Build a groups configuration with a 5-second actor call timeout.

delete

pub fn delete(
  Groups,
  String
) -> Result(Nil, GroupError)

Delete a group.

Returns GroupCallFailed if the actor is unavailable or does not reply within the configured call timeout (5 seconds by default).

list_groups

pub fn list_groups(Groups) -> Result(List(String), GroupError)

Return all group names.

Returns GroupCallFailed if the actor is unavailable or does not reply within the configured call timeout (5 seconds by default).

remove

pub fn remove(
  Groups,
  String,
  String
) -> Result(Nil, GroupError)

Remove a topic from a group.

Returns GroupCallFailed if the actor is unavailable or does not reply within the configured call timeout (5 seconds by default).

topics

pub fn topics(
  Groups,
  String
) -> Result(set.Set(String), GroupError)

Return all topics in a group.

Returns GroupCallFailed if the actor is unavailable or does not reply within the configured call timeout (5 seconds by default).

with_call_timeout

pub fn with_call_timeout(
  Config,
  Int
) -> Config

Set the timeout for synchronous group operations, in milliseconds.

This applies to create, delete, add, remove, topics, and list_groups. These functions return GroupCallFailed if no reply arrives in time. The caller retains no monitor and receives no late reply. A timed-out mutation may still be applied.