Closes the connections a sink adapter opened and then stopped using.
The broker adapters (ankusa_rabbitmq, ankusa_kafka, ankusa_nats,
ankusa_redis) start one connection process per broker on the first
delivery, under their own DynamicSupervisor. Without this, a source that is
deleted, or re-pointed at another broker, leaves its connection open for the
life of the node. Each adapter's Ankusa.Sink.Connections tree runs one
reaper next to its supervisor, and Ankusa.Sink.Connections.ensure/5 calls
touch/4 before every publish; a connection not touched for its idle_ms
is stopped through that supervisor (DynamicSupervisor.terminate_child/2,
so a :permanent child is removed, not restarted), and the next delivery
that needs it starts a new one.
touch/4 is one :ets.insert/2 into a public table, so the publish path
never waits on this process. A sweep runs every :tick_ms (default 30 s); a
row whose connection already died is dropped without a word.
The race it accepts
A delivery that looked its connection up just before the sweep stopped it
fails that attempt (the call exits, which Ankusa.Sink turns into
{:error, _}), and the retry policy's next attempt starts a fresh
connection. A connection touched while the sweep looks at it is kept: the
sweep deletes only the exact row it judged idle. Because the touch comes
before the publish, a publish in flight is never older than idle_ms as long
as idle_ms is longer than the publish can take; idle_ms/2 enforces that
floor for each adapter.
start_link/1 opts: :name (the reaper's registered name and its table's),
:supervisor (the DynamicSupervisor whose children it stops), :tick_ms.
Summary
Functions
An adapter's idle timeout from its sink opts: :idle_timeout_ms (default 10
minutes), 0 for never, else at least floor_ms — the longest one publish
can take, plus a margin — so a sweep never stops a connection under a
publish. Raises ArgumentError on anything but a non-negative integer.
Record that the connection pid, known as key, is about to be used, and
that it may be stopped once unused for idle_ms. idle_ms of 0 means
never: the row is dropped. key is printed when the connection is closed,
so it must not carry a credential.
Functions
@spec idle_ms(keyword(), pos_integer()) :: non_neg_integer()
An adapter's idle timeout from its sink opts: :idle_timeout_ms (default 10
minutes), 0 for never, else at least floor_ms — the longest one publish
can take, plus a margin — so a sweep never stops a connection under a
publish. Raises ArgumentError on anything but a non-negative integer.
@spec touch(atom(), term(), pid(), non_neg_integer()) :: :ok
Record that the connection pid, known as key, is about to be used, and
that it may be stopped once unused for idle_ms. idle_ms of 0 means
never: the row is dropped. key is printed when the connection is closed,
so it must not carry a credential.