Streaming follow-ups from the #3 review

CHANGELOG.md

@@ -18,9 +18,14 @@
       `health?/2`; parsed events (`Hunter.Streaming.Event` — payloads
       decode to `Status`, `Notification`, `Conversation`, `Announcement`,
       `Announcement.Reaction`, or id strings for deletes) are delivered to
-      a subscriber pid as `{:hunter_stream, pid, event}` messages. No
-      automatic reconnection: the process notifies the subscriber and
-      exits, so callers supervise it
+      a subscriber pid as `{:hunter_stream, pid, event}` messages. By
+      default there is no automatic reconnection (the process notifies
+      the subscriber and exits, so callers supervise it); an opt-in
+      `reconnect` mode retries dropped connections with exponential
+      backoff and replays the subscription set ([#142]). Frame-level
+      decode errors tear the connection down instead of skipping the
+      desynced byte stream, and `health?/2` is also reachable as
+      `Hunter.streaming_health?/2` on the facade ([#142])
     - Account extras ([#124]): `lookup_account/2`, `accounts_by_ids/2`,
       `familiar_followers/2` (new `Hunter.FamiliarFollowers` entity),
       `account_featured_tags/2`, `register_account/2` (returns a
@@ -190,6 +195,7 @@
 [#116]: https://github.com/milmazz/hunter/issues/116
 [#122]: https://github.com/milmazz/hunter/issues/122
 [#3]: https://github.com/milmazz/hunter/issues/3
+[#142]: https://github.com/milmazz/hunter/issues/142
 [#139]: https://github.com/milmazz/hunter/issues/139
 [#123]: https://github.com/milmazz/hunter/issues/123
 [#124]: https://github.com/milmazz/hunter/issues/124

README.md

@@ -495,6 +495,27 @@ iex> Hunter.update_credentials(conn, %{note: "Enum.random(~w(programming cycling
 
 Returns a `Hunter.Account`
 
+### Streaming
+
+Real-time events are delivered over Mastodon's multiplexed streaming
+WebSocket. `Hunter.Streaming.connect/2` opens a connection process linked
+to the caller and sends parsed events to the subscriber's mailbox:
+
+```elixir
+iex> {:ok, pid} = Hunter.Streaming.connect(conn, streams: ["user", {"hashtag", tag: "elixir"}])
+{:ok, #PID<0.233.0>}
+iex> flush()
+{:hunter_stream, #PID<0.233.0>,
+ %Hunter.Streaming.Event{stream: ["user"], type: "update", payload: %Hunter.Status{...}}}
+```
+
+Streams can also be joined and left at runtime with
+`Hunter.Streaming.subscribe/3` and `Hunter.Streaming.unsubscribe/3`, and
+the connection closed with `Hunter.Streaming.close/1`. A single
+`{:hunter_stream, pid, {:closed, reason}}` message signals that the socket
+is gone; pass `reconnect: true` to `connect/2` to retry drops with
+exponential backoff instead. See the `Hunter.Streaming` docs for details.
+
 ### Configuration
 
 Hunter uses [Req](https://hex.pm/packages/req) as its HTTP client layer.

lib/hunter.ex

@@ -2481,6 +2481,21 @@ defmodule Hunter do
     Request.request!(conn, :delete, "/api/v2/filters/statuses/#{id}", :empty)
   end
 
+  @doc """
+  Checks the streaming server's health endpoint (Mastodon 2.5+)
+
+  ## Parameters
+
+    * `conn` - connection credentials
+    * `opts` - `url:` overrides the streaming base URL (`ws://`/`wss://`
+      accepted and mapped to `http://`/`https://`)
+
+  See `Hunter.Streaming.health?/2`; streaming connections themselves are
+  opened with `Hunter.Streaming.connect/2`.
+  """
+  @spec streaming_health?(Hunter.Client.t(), Keyword.t()) :: boolean
+  defdelegate streaming_health?(conn, opts \\ []), to: Hunter.Streaming, as: :health?
+
   @doc """
   Returns Hunter version
   """

lib/hunter/streaming.ex

@@ -6,8 +6,17 @@ defmodule Hunter.Streaming do
   events arrive in the subscriber's mailbox as
   `{:hunter_stream, connection_pid, %Hunter.Streaming.Event{}}` and a
   single `{:hunter_stream, connection_pid, {:closed, reason}}` is sent
-  when the socket closes. There is no automatic reconnection: supervise
-  and restart the connection from the consuming application.
+  when the socket closes. By default there is no automatic reconnection:
+  supervise and restart the connection from the consuming application.
+
+  With the `reconnect` option, a drop of an established connection is
+  retried with exponential backoff instead of being terminal: the
+  subscriber receives `{:hunter_stream, connection_pid, {:reconnecting,
+  reason}}` when the socket drops and `{:hunter_stream, connection_pid,
+  :reconnected}` once it is re-established and the current subscription
+  set has been replayed. `close/1` remains terminal, and a reconnect mode
+  with `max_attempts` still delivers `{:closed, reason}` and exits once
+  the attempts are exhausted.
 
   Instances may serve streaming from a different host than the REST API;
   discover it via `Hunter.instance_info/1` under
@@ -36,6 +45,16 @@ defmodule Hunter.Streaming do
       (see the module docs for discovery)
     * `transport_opts` - Mint transport options, e.g.
       `[verify: :verify_none]` for self-signed certificates
+    * `reconnect` - `true` or a keyword list to reconnect automatically
+      when an established connection drops (see the module docs for the
+      subscriber messages involved), default: `false`. The initial
+      connection is always synchronous: `connect/2` still returns
+      `{:error, reason}` when it fails. Keys:
+      * `initial_backoff` - delay in ms before the first attempt,
+        doubled on each failure, default: `1_000`
+      * `max_backoff` - backoff ceiling in ms, default: `30_000`
+      * `max_attempts` - consecutive failed attempts tolerated before
+        giving up, default: `:infinity`
 
   """
   @spec connect(Hunter.Client.t(), Keyword.t()) :: {:ok, pid} | {:error, term}
@@ -44,7 +63,8 @@ defmodule Hunter.Streaming do
       uri: ws_uri(conn, opts),
       subscriber: Keyword.get(opts, :subscriber, self()),
       streams: Keyword.get(opts, :streams, []),
-      transport_opts: Keyword.get(opts, :transport_opts, [])
+      transport_opts: Keyword.get(opts, :transport_opts, []),
+      reconnect: Keyword.get(opts, :reconnect, false)
     )
   end