ExNtfy.Subscription (ex_ntfy v0.1.1)

Copy Markdown View Source

A long-lived subscription to ntfy topics — a GenServer owning one streaming HTTP connection, with automatic reconnect (resuming via since=<last id>), a keepalive watchdog, and two consumption styles.

Usually started through ExNtfy.subscribe/2; use start_link/1 directly in a supervision tree:

children = [
  {ExNtfy.Subscription,
   topics: "alerts", handler: {MyHandler, []}, name: MyApp.NtfySub}
]

In message-passing mode the owner receives {:ntfy, pid, %ExNtfy.Message{}} for messages and {:ntfy_lifecycle, pid, event} for :connected, :disconnected, {:down, reason}, and {:message_clear | :message_delete, message}. open/keepalive events are internal and never surface.

Alongside the options below, all ExNtfy.Subscribe.Options (filters, since, scheduled — but not poll) and ExNtfy.Config client options are accepted. Req's own retry is disabled on the stream request — this module's reconnect loop is in charge.

format: :ws subscribes over WebSocket (GET /<topics>/ws) with identical semantics — it needs the optional :mint_web_socket dependency in your deps, and subscribe/2 raises a clear ArgumentError without it. See ExNtfy.Stream.WS.

A non-2xx response is treated as fatal (it won't fix itself by retrying — think 403): the subscription delivers {:down, %ExNtfy.Error{}} and stops with {:shutdown, error} regardless of reconnect:.

Telemetry

[:ex_ntfy, :subscription, :connected | :disconnected | :message] with metadata %{topics: topics} — never credentials or message contents.

Options

  • :topics - Required. Topic(s) to subscribe to — a string, list, or comma-separated string.

  • :format - Stream format. :json is the primary; :raw carries bodies only (no metadata, so no since resume); :ws subscribes over WebSocket and requires the optional :mint_web_socket dependency. The default value is :json.

  • :handler - {module, init_arg} implementing ExNtfy.Handler (handler mode).

  • :owner (pid/0) - Receiver of {:ntfy, pid, message} / {:ntfy_lifecycle, pid, event} (message-passing mode; defaults to the caller). The subscription stops when the owner dies.

  • :reconnect (boolean/0) - Reconnect with backoff on disconnect. false stops instead. The default value is true.

  • :reconnect_base_ms (pos_integer/0) - First backoff delay; doubles per attempt (plus jitter). The default value is 1000.

  • :reconnect_max_ms (pos_integer/0) - Backoff cap. The default value is 60000.

  • :idle_timeout (pos_integer/0) - Watchdog: with no stream activity for this long (server keepalives arrive ~every 45 s), the connection is considered dead and rebuilt. The default value is 90000.

  • :name (term/0) - Optional GenServer name, for supervision trees.

Summary

Functions

Returns a specification to start this module under a supervisor.

Starts a subscription linked to the caller. Options as described in the moduledoc; invalid options raise in the caller.

A lazy Enumerable of ExNtfy.Message structs — blocks the calling process. The underlying subscription starts on first demand and stops when the consumer halts (e.g. Enum.take/2) or exits.

Starts a subscription to topics. Sugar over start_link/1 that defaults owner: to the caller in message-passing mode.

Stops a subscription cleanly.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts a subscription linked to the caller. Options as described in the moduledoc; invalid options raise in the caller.

stream(topics, opts \\ [])

A lazy Enumerable of ExNtfy.Message structs — blocks the calling process. The underlying subscription starts on first demand and stops when the consumer halts (e.g. Enum.take/2) or exits.

subscribe(topics, opts \\ [])

Starts a subscription to topics. Sugar over start_link/1 that defaults owner: to the caller in message-passing mode.

unsubscribe(server)

@spec unsubscribe(GenServer.server()) :: :ok

Stops a subscription cleanly.