PubSub

PubSub module: topic-based publish-subscribe backbone for Channels.

PubSub maintains a mapping from topic strings to sets of subscriber pids. Channels subscribe on join and unsubscribe on leave; broadcasts fan out to all subscribers of a topic.

Two layers: Pure layer — PubSubState + subscribe_state/unsubscribe_state/get_subscribers. Fully testable without the actor runtime. Runtime API — subscribe/unsubscribe/broadcast/broadcast_from. Requires the actor system (spawn, actor_cast).

Sharding: topics are routed to a shard by hashing the topic string. topic_shard(topic, n) → 0..n-1 uses the built-in hash() for simplicity. Replace with an explicit FNV-1a loop if byte-level control is needed.

Wire message sent to subscribers on broadcast: PubSubBroadcast(topic, event, payload_string)

Usage (pure): let s = PubSub.new_state() let s1 = PubSub.subscribe_state(s, "room:lobby", my_pid) let subs = PubSub.get_subscribers(s1, "room:lobby") -- [my_pid] let s2 = PubSub.unsubscribe_state(s1, "room:lobby", my_pid) let subs2 = PubSub.get_subscribers(s2, "room:lobby") -- []

Types

typePubSubStatePubSubState = PubSubState(Map(String, List(Int)))#

Immutable PubSub state: Map(topic, List(subscriber_pid)).

Functions

fnbroadcast_in_shardbroadcast_in_shard(shard_pid, topic, excluded_pid, event, payload)#

Broadcast event + payload to all subscribers of topic in shard shard_pid. Pass Some(sender_pid) to exclude the sender (broadcast_from behaviour), None to reach everyone. Sends a PubSubBroadcast message to the shard actor (fire and forget). Requires actor runtime.

PubSub.broadcast_in_shard(shard_pid, "room:lobby", None, "new_msg", "{}")

fnbroadcast_tobroadcast_to(pids, excluded_pid, event, payload)#

Send a broadcast message to all subscribers in pids except excluded_pid. When excluded_pid is None, all subscribers receive the message. Uses actor_cast to deliver BroadcastMsg(event, payload) to each pid, in list order; a pid listed twice receives it twice. Returns (). Requires actor runtime.

PubSub.broadcast_to(PubSub.get_subscribers(state, topic), Some(self()), "new_msg", payload)

fnget_subscribersget_subscribers(ps, topic)#

The subscriber pids for topic, newest first (duplicates included). Empty list if none. Topics are matched exactly; patterns are not expanded.

Examples

march> PubSub.get_subscribers(PubSub.new_state(), "room:*")
[]

fnhas_subscribershas_subscribers(ps, topic)#

True when at least one pid is subscribed to topic.

Examples

march> PubSub.has_subscribers(PubSub.new_state(), "room:lobby")
false

fnnew_statenew_state()#

An empty PubSub state: no topic has subscribers.

Examples

march> PubSub.get_subscribers(PubSub.new_state(), "room:lobby")
[]

fnsubscribe_in_shardsubscribe_in_shard(shard_pid, topic, subscriber_pid)#

Subscribe subscriber_pid to topic in the shard shard_pid. Sends a PubSubSubscribe message to the shard actor (fire and forget). Requires actor runtime.

PubSub.subscribe_in_shard(ChannelServer.select_shard(shards, topic), topic, self())

fnsubscribe_statesubscribe_state(ps, topic, pid)#

Subscribe pid to topic. Returns the updated state. The pid is prepended, so get_subscribers lists the newest subscriber first. Subscribing a pid that is already subscribed adds a DUPLICATE entry: nothing deduplicates at send time, so broadcast_to then delivers to it twice, and it takes two unsubscribe_state calls to remove.

let s = PubSub.subscribe_state(PubSub.new_state(), "room:lobby", 7)
PubSub.get_subscribers(s, "room:lobby")   -- [7]

fnsubscriber_countsubscriber_count(ps, topic)#

Number of subscription entries for topic (a pid subscribed twice counts twice). O(n).

Examples

march> PubSub.subscriber_count(PubSub.new_state(), "room:lobby")
0

fntopic_matchestopic_matches(pattern, topic)#

True when pattern matches topic. Segment-aware: a trailing * in pattern wildcards the remainder, but the character at the wildcard position in topic must be a segment separator (: or .) or end-of-string. This prevents "user*" from matching "users" or "user_other" — Phoenix's pubsub convention.

A bare "*" matches every topic, including ""; an empty pattern matches only the empty topic. A * anywhere but the end is a literal character.

Examples

march> PubSub.topic_matches("room:*", "room:lobby")
true
march> PubSub.topic_matches("room:*", "room:1:x")
true
march> PubSub.topic_matches("room:*", "game:42")
false
march> PubSub.topic_matches("room:lobby", "room:lobby")
true
march> PubSub.topic_matches("user*", "user:1")
true
march> PubSub.topic_matches("user*", "user")
true
march> PubSub.topic_matches("user*", "users")
false
march> PubSub.topic_matches("*", "")
true

fntopic_shardtopic_shard(topic, num_shards)#

Hash-based shard index for topic given num_shards shards. Uses the built-in hash() function (same primitive the HAMT uses), masked to its low 31 bits. Returns a value in [0, num_shards). The index depends on the platform's hash, so do not persist it or share it between nodes built for different targets. Panics (division by zero) if num_shards is 0.

Examples

march> PubSub.topic_shard("room:lobby", 1)
0

fnunsubscribe_in_shardunsubscribe_in_shard(shard_pid, topic, subscriber_pid)#

Unsubscribe subscriber_pid from topic in the shard shard_pid. Sends a PubSubUnsubscribe message to the shard actor (fire and forget). Requires actor runtime.

PubSub.unsubscribe_in_shard(shard_pid, "room:lobby", self())

fnunsubscribe_stateunsubscribe_state(ps, topic, pid)#

Unsubscribe pid from topic, removing only its first (newest) entry. When no subscriber is left the topic is dropped from the state. A pid or topic that is not subscribed leaves the state unchanged.

let s = PubSub.subscribe_state(PubSub.new_state(), "room:lobby", 7)
PubSub.has_subscribers(PubSub.unsubscribe_state(s, "room:lobby", 7), "room:lobby")   -- false

Examples

march> PubSub.get_subscribers(PubSub.unsubscribe_state(PubSub.new_state(), "room:lobby", 7), "room:lobby")
[]