brazier - v0.1.0

brazier/pubsub

Types

PubSubConfig

</>

PubSub configuration. Use default_config and the setters.

pub type PubSubConfig {
  PubSubConfig(
    partitions: Int,
    pool_size: Int,
    max_forward_queue: Int,
    max_topics_per_subscriber: Int,
  )
}

Constructors 

  • PubSubConfig(
  partitions: Int,
  pool_size: Int,
  max_forward_queue: Int,
  max_topics_per_subscriber: Int,
)

    Arguments

    partitions

    Registry partitions on this node. Defaults to the number of online schedulers.

    pool_size

    Forwarding workers per node. Must be equal on every node.

    max_forward_queue

    Queue length above which a worker reports overload.

    max_topics_per_subscriber

    Topics a single process may subscribe to.

PubSubError

</>

Why a PubSub operation failed.

pub type PubSubError {
  InvalidTopic(String)
  NotStarted
  NodeNotConnected(node: String)
  NodeNotVerified(node: String)
  TooManySubscriptions(limit: Int)
  InvalidConfig(String)
  StartFailed(String)
  StopFailed(String)
}

Constructors 

  • InvalidTopic(String)

    The topic is empty or longer than 255 bytes.

  • NotStarted

    PubSub is not running on this node (or a registry partition is restarting).

  • NodeNotConnected(node: String)

    direct_broadcast targeted a node that is not connected or runs no PubSub worker for this shard.

  • NodeNotVerified(node: String)

    direct_broadcast targeted a connected node that has not passed the cluster peer check (or no cluster manager runs here).

  • TooManySubscriptions(limit: Int)

    The calling process already holds the maximum number of topics.

  • InvalidConfig(String)

    The configuration is invalid, or differs from the running one.

  • StartFailed(String)

    The PubSub supervisor did not start.

  • StopFailed(String)

    The PubSub supervisor did not shut down within the stop deadline.

Topic

</>

A topic for pub/sub messaging: 1 to 255 bytes.

pub type Topic =
  String

Values

broadcast

</> 
pub fn broadcast(topic: String, message: msg) -> Nil

Broadcast to every subscriber on every cluster member. Crashes on any error.

default_config

</> 
pub fn default_config() -> PubSubConfig

The default configuration: one partition per online scheduler, one forwarding worker, overload reported above 10,000 queued forwards, at most 1,000 topics per subscriber.

describe

</> 
pub fn describe(error: PubSubError) -> String

A readable description of an error.

direct_broadcast

</> 
pub fn direct_broadcast(
  target: node.Node,
  topic: String,
  message: msg,
) -> Result(Nil, PubSubError)

Deliver message to the subscribers of topic on target only. Returns Error(NodeNotConnected(_)) if target is not connected, and Error(NodeNotVerified(_)) if it has not passed the cluster peer check.

local_broadcast

</> 
pub fn local_broadcast(
  topic: String,
  message: msg,
) -> Result(Nil, PubSubError)

Deliver message to the subscribers of topic on this node only.

local_subscriber_count

</> 
pub fn local_subscriber_count(topic: String) -> Int

The number of processes on this node subscribed to topic. Zero when PubSub is not running, because then nothing can be subscribed.

local_subscribers

</> 
pub fn local_subscribers(
  topic: String,
) -> Result(List(process.Pid), PubSubError)

The processes on this node subscribed to topic.

max_forward_queue

</> 
pub fn max_forward_queue(
  config: PubSubConfig,
  limit: Int,
) -> PubSubConfig

Set the forwarding queue length that triggers an overload report.

max_partitions

</> 
pub const max_partitions: Int

Upper bound for partitions.

max_pool_size

</> 
pub const max_pool_size: Int

Upper bound for pool_size.

max_topic_bytes

</> 
pub const max_topic_bytes: Int

The longest topic, in bytes.

max_topics_limit

</> 
pub const max_topics_limit: Int

Upper bound for max_topics_per_subscriber.

max_topics_per_subscriber

</> 
pub fn max_topics_per_subscriber(
  config: PubSubConfig,
  limit: Int,
) -> PubSubConfig

Set how many topics one process may subscribe to (1 to 100,000).

partition_for

</> 
pub fn partition_for(
  pid: process.Pid,
) -> Result(process.Pid, PubSubError)

The registry partition that holds pid’s subscriptions. A subscriber is linked to it; monitoring it as well lets a process that traps exits notice when the registry dies.

partitions

</> 
pub fn partitions(
  config: PubSubConfig,
  partitions: Int,
) -> PubSubConfig

Set the number of registry partitions (1 to 1024).

pool_size

</> 
pub fn pool_size(
  config: PubSubConfig,
  pool_size: Int,
) -> PubSubConfig

Set the number of forwarding workers per node (1 to 64).

running_config

</> 
pub fn running_config() -> Result(PubSubConfig, PubSubError)

The configuration PubSub runs with on this node.

start

</> 
pub fn start() -> Nil

Make sure PubSub runs on this node, with the default configuration if it is not running yet. It is not linked to the caller.

Crashes, after logging the reason, if PubSub cannot start.

start_with

</> 
pub fn start_with(
  config: PubSubConfig,
) -> Result(Nil, PubSubError)

Start PubSub with config, not linked to the caller.

Returns Ok(Nil) if it is already running with the same configuration and Error(InvalidConfig(_)) if it runs with a different one.

stop

</> 
pub fn stop() -> Result(Nil, PubSubError)

Stop PubSub on this node (any process may call it). Every subscriber is linked to its partition, so subscribers stop too. Use it for PubSub started with start or start_with; under supervised, stop the parent supervisor instead, or it restarts PubSub.

subscribe

</> 
pub fn subscribe(topic: String) -> Nil

Subscribe the current process to a topic. Crashes on any error.

subscriber_count

</> 
pub fn subscriber_count(topic: String) -> Int

The number of subscribers of topic on this node.

Subscriptions are node-local, so this equals local_subscriber_count.

supervised

</> 
pub fn supervised(
  config: PubSubConfig,
) -> Result(supervision.ChildSpecification(Nil), PubSubError)

A child specification that starts PubSub linked to a supervisor. This is the form the root supervision tree uses.

try_broadcast

</> 
pub fn try_broadcast(
  topic: String,
  message: msg,
) -> Result(Nil, PubSubError)

Deliver message to every subscriber of topic on this node and on every cluster member (connected nodes that passed the peer check).

try_subscribe

</> 
pub fn try_subscribe(topic: String) -> Result(Nil, PubSubError)

Subscribe the calling process to topic. Subscribing twice to the same topic is one subscription: each broadcast is delivered once.

try_unsubscribe

</> 
pub fn try_unsubscribe(topic: String) -> Result(Nil, PubSubError)

Unsubscribe the calling process from topic. Unsubscribing from a topic the process is not subscribed to is Ok.

unsubscribe

</> 
pub fn unsubscribe(topic: String) -> Nil

Unsubscribe the current process from a topic. Crashes on any error.

validate_config

</> 
pub fn validate_config(
  config: PubSubConfig,
) -> Result(PubSubConfig, PubSubError)

Check every field of a configuration.

validate_topic

</> 
pub fn validate_topic(topic: String) -> Result(Nil, PubSubError)

Check that topic is a valid topic: 1 to 255 bytes.

✨ Search Document