ExWapp.Session.Worker.Subscribers (ExWapp v0.1.2)

Copy Markdown View Source

Subscriber bookkeeping for QR and message streams.

Summary

Functions

Broadcasts one call-log state change parsed from an inbound <call> stanza.

Broadcasts the server's verdict on an outbound message.

Broadcasts a session error via lifecycle PubSub topics.

Broadcasts session health events (errors, drops, retries, guard trips).

Broadcasts a session lifecycle event via PubSub.

Broadcasts a transient reconnect notice via lifecycle PubSub topics.

Functions

broadcast_call(session_id, call)

@spec broadcast_call(term(), ExWapp.Call.t()) :: :ok

Broadcasts one call-log state change parsed from an inbound <call> stanza.

Published because only the session observes call signalling; without it a caller would have to poll ExWapp.Call.list/2 to notice a ringing or missed call. Every phase change ships the merged %ExWapp.Call{}, so subscribers read the current status from the struct itself.

Events are sent on:

  • per-session: "ex_wapp:calls:<session_id>" with {:ex_wapp_call, call}
  • global: "ex_wapp:calls" with {session_id, {:ex_wapp_call, call}}

broadcast_delivery(session_id, message_id, status, meta)

@spec broadcast_delivery(term(), String.t(), atom(), map()) :: :ok

Broadcasts the server's verdict on an outbound message.

WhatsApp answers a message it accepts with an <ack> and stays silent for one it drops, so this is the only positive confirmation a sender ever gets. It is published because the session already observes it while the caller cannot: without it the only way to learn a send succeeded is to poll stored message status, which is both slower and unable to distinguish "refused" from "no answer yet".

Events are sent on:

  • per-session: "ex_wapp:delivery:<session_id>"
  • global: "ex_wapp:delivery" with {session_id, payload}

Message format:

  • {:ex_wapp_delivery, message_id, :sent, %{jid: jid}}
  • {:ex_wapp_delivery, message_id, :failed, %{jid: jid, error: %ExWapp.Error.Ack{}}}

broadcast_error(session_id, reason)

@spec broadcast_error(term(), term()) :: :ok

Broadcasts a session error via lifecycle PubSub topics.

Message format:

  • {:ex_wapp_session, :error, reason}

broadcast_health(session_id, event)

@spec broadcast_health(term(), term()) :: :ok

Broadcasts session health events (errors, drops, retries, guard trips).

Events are sent on:

  • per-session: "ex_wapp:health:<session_id>"
  • global: "ex_wapp:health" with {session_id, payload}

broadcast_lifecycle(session_id, event)

@spec broadcast_lifecycle(term(), term()) :: :ok

Broadcasts a session lifecycle event via PubSub.

Events are sent on two topics:

  • Per-session: "ex_wapp:session:<session_id>" with payload as-is
  • Global: "ex_wapp:session" with {session_id, payload}

broadcast_reconnecting(session_id, reason)

@spec broadcast_reconnecting(term(), term()) :: :ok

Broadcasts a transient reconnect notice via lifecycle PubSub topics.

Emitted for recoverable server-side conditions (e.g. 503/515 stream errors) so owners can stay calm instead of treating them as a hard error.

Message format:

  • {:ex_wapp_session, :reconnecting, reason}

cleanup_dead_subscriber(state, dead_mon)

@spec cleanup_dead_subscriber(ExWapp.Session.Worker.t(), reference()) ::
  ExWapp.Session.Worker.t()

notify_message_subscribers(state, jid, message)

@spec notify_message_subscribers(ExWapp.Session.Worker.t(), String.t(), map()) :: :ok

notify_qr(state, event, emit_telemetry_fun)

@spec notify_qr(ExWapp.Session.Worker.t(), term(), (-> any())) :: :ok

remove_message_subscriber(state, ref)

@spec remove_message_subscriber(ExWapp.Session.Worker.t(), reference()) ::
  ExWapp.Session.Worker.t()

remove_qr_subscriber(state, ref)

@spec remove_qr_subscriber(ExWapp.Session.Worker.t(), reference()) ::
  ExWapp.Session.Worker.t()