Amarula.Connection.AckLifecycle (amarula v0.4.2)

View Source

The parked-send / sender-monitor seam for Amarula.Connection — shared by the send path, the receive <ack> handler, and the sender :DOWN/:ack_timeout handlers.

A send parks the consumer's from under its msg_id (in state.pending_acks, keyed {from, on_ack, timer, jid}) and is answered only when the server's <ack> arrives, the send fails, the timeout fires, or the recipient's sender crashes. One monitor per recipient (state.sender_monitors) covers all of that recipient's in-flight sends.

This module owns those two maps' transformations. It is a state module (like Socket.IQ), not pure-pure: it threads state and performs side effects on stored refs/froms (Process.cancel_timer, Process.demonitor, GenServer.reply) — none of which need the connection's own pid. Creating a timer/monitor (which needs self()) stays on Connection, which passes the ready timer/ref in.

Summary

Types

state()

Functions

drop_monitor_if_idle(state, jid)

Drop jid's monitor once it has no remaining parked sends, so monitor refs don't leak and we don't hold a stale monitor on a sender that will idle-stop.

fail_recipient_sends(state, jid, reason)

Fail every parked send for a crashed recipient with {:error, {:sender_crashed, reason}}, cancelling their timers. (The monitor entry was already removed by pop_monitor_by_ref/2.)

monitored?(state, jid)

True once we already monitor jid's sender — the caller then skips creating one.

park(state, msg_id, from, timer, on_ack, jid)

Park a send. from nil = fire-and-forget (nothing parked: no caller waits, so a missing ack is fine). Otherwise store {from, on_ack, timer, jid} under msg_id, where on_ack applies the success shape. timer is created by the caller (it needs self()).

pop_monitor_by_ref(state, ref)

Find which recipient a :DOWN ref belonged to and drop that monitor entry, returning {jid | nil, state}. The monitor already fired, so no demonitor.

put_monitor(state, jid, ref)

Record a recipient's sender monitor ref (first parked send only).

resolve(state, msg_id, reply_fun)

Resolve a parked send: reply reply_fun.(on_ack) to the caller, cancel its timeout timer, drop the entry, and stop monitoring the recipient once it has no more parked sends. An unknown msg_id (already resolved, fire-and-forget, or not ours) is a no-op — the same id never resolves twice.

timeout_ms(arg1)

The ack timeout, overridable via config[:ack_timeout_ms] (tests).

Types

state()

@type state() :: map()

Functions

drop_monitor_if_idle(state, jid)

@spec drop_monitor_if_idle(state(), String.t()) :: state()

Drop jid's monitor once it has no remaining parked sends, so monitor refs don't leak and we don't hold a stale monitor on a sender that will idle-stop.

fail_recipient_sends(state, jid, reason)

@spec fail_recipient_sends(state(), String.t(), term()) :: state()

Fail every parked send for a crashed recipient with {:error, {:sender_crashed, reason}}, cancelling their timers. (The monitor entry was already removed by pop_monitor_by_ref/2.)

monitored?(state, jid)

@spec monitored?(state(), String.t()) :: boolean()

True once we already monitor jid's sender — the caller then skips creating one.

park(state, msg_id, from, timer, on_ack, jid)

@spec park(
  state(),
  String.t(),
  GenServer.from() | nil,
  reference(),
  (term() -> term()),
  String.t()
) :: state()

Park a send. from nil = fire-and-forget (nothing parked: no caller waits, so a missing ack is fine). Otherwise store {from, on_ack, timer, jid} under msg_id, where on_ack applies the success shape. timer is created by the caller (it needs self()).

pop_monitor_by_ref(state, ref)

@spec pop_monitor_by_ref(state(), reference()) :: {String.t() | nil, state()}

Find which recipient a :DOWN ref belonged to and drop that monitor entry, returning {jid | nil, state}. The monitor already fired, so no demonitor.

put_monitor(state, jid, ref)

@spec put_monitor(state(), String.t(), reference()) :: state()

Record a recipient's sender monitor ref (first parked send only).

resolve(state, msg_id, reply_fun)

@spec resolve(state(), String.t(), ((term() -> term()) -> term())) :: state()

Resolve a parked send: reply reply_fun.(on_ack) to the caller, cancel its timeout timer, drop the entry, and stop monitoring the recipient once it has no more parked sends. An unknown msg_id (already resolved, fire-and-forget, or not ours) is a no-op — the same id never resolves twice.

timeout_ms(arg1)

@spec timeout_ms(state()) :: pos_integer()

The ack timeout, overridable via config[:ack_timeout_ms] (tests).