The broker connections a sink adapter opens on demand. An adapter package
starts one tree with start_link/1 from its Application: a
DynamicSupervisor (<sink>.Supervisor) that owns the connections and an
Ankusa.Sink.Reaper (<sink>.Reaper) that closes idle ones, under
<sink>.TopSup. A sink calls ensure/5 before every publish.
The connections live in the adapter's tree, not the instance's, so a lost
broker never takes ingest or dispatch down: it makes publish/4 (see
Ankusa.Sink.Messaging) return {:error, _} for the source's
Ankusa.RetryPolicy. A connection that belongs to one instance registers
through that instance's registry (Ankusa.via/2, found again with
Ankusa.whereis/2), so nothing is registered globally and two instances
never share one; core itself does not know which adapters are loaded.
Summary
Functions
The connection key names, started on demand and about to be used.
The reaper (and its table) that closes sink's idle connections.
Start sink's connection tree; an adapter's Application.start/2 returns this.
The DynamicSupervisor that owns sink's connections.
Functions
@spec ensure( module(), term(), (-> pid() | nil), (-> Supervisor.child_spec()), non_neg_integer() ) :: {:ok, pid()} | {:error, term()}
The connection key names, started on demand and about to be used.
lookup finds the running one (Ankusa.whereis/2, Process.whereis/1);
nil starts child.() under sink's supervisor — only then is the child
spec built, and a start that loses the race ({:error, {:already_started, pid}}) is that pid. The pid is touched in sink's reaper under key with
idle_ms (Ankusa.Sink.Reaper.touch/4; 0 never reaps), so key is
printed when the connection is closed and must carry no credential.
The reaper (and its table) that closes sink's idle connections.
@spec start_link(module()) :: Supervisor.on_start()
Start sink's connection tree; an adapter's Application.start/2 returns this.
The DynamicSupervisor that owns sink's connections.