Direct, monitored process subscriptions for GenServers that publish events.

This is the producer side of the Sagents publish/subscribe transport. It provides per-server event channels (AgentServer, FileSystemServer) with a point-to-point send/2 between the producer GenServer and a small set of subscriber pids it tracks itself.

Why not Phoenix.PubSub?

Phoenix.PubSub is a cluster-aware broadcast bus. With a clustered adapter every published message is serialized and shipped to every node, where each node's local PubSub server filters by topic. For per-agent event channels this is a lot of cross-node noise for what is, at the receiver, almost always a single in-node process.

Direct send/2 works transparently across nodes when the pid is remote, but routes one hop per known subscriber rather than a per-node broadcast.

Producer side

A producer GenServer embeds a Sagents.Publisher.State struct in its own state and uses this module. The use macro installs the handle_call/3 clauses that handle subscribe/unsubscribe.

defmodule MyServer do
  use GenServer
  use Sagents.Publisher,
    channels: [:main, :debug],
    state_field: :publisher

  defstruct [..., publisher: Sagents.Publisher.State.new([:main, :debug])]

  def init(opts) do
    {:ok, %__MODULE__{publisher: Sagents.Publisher.State.new([:main, :debug])}}
  end

  # Inside any handler that needs to broadcast:
  defp emit(state, event) do
    Sagents.Publisher.broadcast(state.publisher, :main, {:my_server, event})
    state
  end
end

Consumer side

Subscribers locate the producer by name (typically via Sagents.ProcessRegistry via_tuple), call Sagents.Publisher.subscribe/3, and receive the wrapped events as ordinary process messages. The producer monitors each subscriber to clean up on death.

Typical entry points

Most consumers should not call Publisher.subscribe/3 directly — there are higher-level wrappers that take an agent id (or filesystem scope key) and resolve the via-tuple internally:

• Sagents.AgentServer.subscribe/3 — subscribe(agent_id, channel \ :main, subscriber_pid \ nil). Covers the common case of subscribing self() to :main, plus :debug (used by sagents_live_debugger) and foreign-pid subscribers (e.g. a bridge GenServer proxying events to another transport).
• Sagents.FileSystemServer.subscribe/1 — same shape for filesystem change events.
• Sagents.Subscriber — for hosts (LiveView, GenServer) that need crash-recovery + Phoenix.Presence-driven re-subscription. Wraps the AgentServer/FileSystemServer entry points and threads a subs map through the host's state.

Reach for Publisher.subscribe/3 directly only when implementing a new producer module that needs to expose its own subscribe/N shorthand.

Returned subscription handle

subscribe/3 returns {:ok, server_pid, monitor_ref}. The subscriber is expected to Process.monitor/1 the server pid (or use the included monitor_ref from this side — the producer's monitor on the subscriber) to detect server death and trigger re-subscription via Presence arrival.

Most consumers should use Sagents.Subscriber rather than calling this module directly.