Sagents.Publisher.State (Sagents v0.16.2)

Subscriber bookkeeping for a Sagents.Publisher producer.

Tracks subscribers across one or more channels and maintains a reverse index from monitor ref → {pid, channel} so {:DOWN, ref, ...} messages can be cleaned up in O(1) without iterating channels.

Subscriber lookup by pid within a channel is also O(1), so duplicate subscribe calls dedupe trivially (returning the existing monitor ref rather than starting a second monitor).

Each subscription also carries a tag/0 recording how its events are addressed. A process holding several subscriptions at once needs the envelope to name the subscription; one that holds a single subscription does not, and pays nothing for the option.

Summary

Types

channel()

A channel identifier.

entry()

One subscription: the producer's monitor on the subscriber, plus its envelope shape.

t()

tag()

How events are addressed to one subscription.

Functions

add(state, channel, pid, tag \\ :untagged)

Add a subscriber pid to a channel with the given envelope shape.

count(state)

Total number of subscriptions across all channels (useful for tests/metrics).

new(channels \\ [:main])

Build a fresh publisher state with the given channel atoms pre-initialized to empty subscriber maps.

remove_pid(state, channel, pid)

Remove a subscriber pid from a channel.

remove_ref(state, ref)

Remove a subscriber by monitor ref (for :DOWN cleanup).

resolve_tag(opts, default_identity)

Resolve subscription options into a tag/0.

seed(state, entries, default_identity \\ nil)

Bulk-seed subscribers from a list of {channel, pid} or {channel, pid, opts} tuples.

subscribed?(state, channel, pid)

Whether a pid is subscribed to a channel.

subscriber_entries(state, channel)

Subscriber entries for a channel as %{pid => entry}.

subscribers(state, channel)

List subscriber pids for a channel.

tag_for(state, channel, pid)

The envelope shape for one subscription. :untagged when the pid is not subscribed to the channel, which is the shape a producer would use anyway.

tag_to_opts(arg1)

Turn a tag/0 back into the option list that produces it.

Types

channel()

@type channel() :: atom()

A channel identifier.

entry()

@type entry() :: %{ref: reference(), tag: tag()}

One subscription: the producer's monitor on the subscriber, plus its envelope shape.

t()

@type t() :: %Sagents.Publisher.State{
  channels: %{required(channel()) => %{required(pid()) => entry()}},
  monitors: %{required(reference()) => {pid(), channel()}}
}

tag()

@type tag() :: :untagged | {:tag, term()}

How events are addressed to one subscription.

:untagged delivers the bare envelope the producer builds for everyone. {:tag, value} delivers an envelope carrying value, so a process holding several subscriptions can attribute each event to one of them.

The value is wrapped rather than stored bare so that nil is a usable tag.

Functions

add(state, channel, pid, tag \\ :untagged)

@spec add(t(), channel(), pid(), tag()) :: {reference(), t()}

Add a subscriber pid to a channel with the given envelope shape.

Idempotent for registration — a pid already subscribed to the channel keeps its monitor and its existing ref is returned. The tag is restated, because the most recent call is the one saying how this mailbox wants its events shaped.

Returns {ref, new_state}.

count(state)

@spec count(t()) :: non_neg_integer()

Total number of subscriptions across all channels (useful for tests/metrics).

new(channels \\ [:main])

@spec new([channel()]) :: t()

Build a fresh publisher state with the given channel atoms pre-initialized to empty subscriber maps.

Channels are auto-created on first subscribe, but pre-declaring them makes the supported set explicit at the host module level.

remove_pid(state, channel, pid)

@spec remove_pid(t(), channel(), pid()) :: t()

Remove a subscriber pid from a channel.

Demonitors the existing monitor (with :flush to drop any in-flight DOWN). No-op if the pid is not subscribed.

remove_ref(state, ref)

@spec remove_ref(t(), reference()) :: {:ok, t()} | :error

Remove a subscriber by monitor ref (for :DOWN cleanup).

Returns {:ok, new_state} if the ref was tracked, :error otherwise.

resolve_tag(opts, default_identity)

@spec resolve_tag(
  keyword(),
  term()
) :: tag()

Resolve subscription options into a tag/0.

tag: value addresses events with the host's own routing key. nil is a legal value.
tagged: true addresses events with default_identity, the identity the producer knows itself by (an agent id, a filesystem scope key).
Neither delivers the bare envelope.

:tag wins when both are given.

Examples

iex> Sagents.Publisher.State.resolve_tag([], "agent-1")
:untagged

iex> Sagents.Publisher.State.resolve_tag([tagged: true], "agent-1")
{:tag, "agent-1"}

iex> Sagents.Publisher.State.resolve_tag([tag: :card_7], "agent-1")
{:tag, :card_7}

iex> Sagents.Publisher.State.resolve_tag([tag: nil], "agent-1")
{:tag, nil}

seed(state, entries, default_identity \\ nil)

@spec seed(t(), [{channel(), pid()} | {channel(), pid(), keyword()}], term()) :: t()

Bulk-seed subscribers from a list of {channel, pid} or {channel, pid, opts} tuples.

Useful in producer init/1 to enroll subscribers passed as :initial_subscribers before the GenServer starts handling messages. This closes the race between "agent started" and "subscriber missed initial events" (e.g., the :status_changed and :node_transferred broadcasts emitted from handle_continue/2).

opts accepts :tag and :tagged, resolved against default_identity by resolve_tag/2, so a seeded subscription is addressed exactly as one established through the producer's own subscribe function.

Each entry establishes a monitor exactly as add/4 would. Duplicates within the list are deduped per-channel — the existing monitor is reused, and the last entry's tag is the one that stands.

Returns the updated state.