Live change-events client for ArcadeDB's /ws WebSocket feed — arcadic's
one caller-supervised process.
Start it under your own supervision tree, then subscribe/3 a database. The
process presents conn.auth on the handshake, keeps the socket drained
continuously, and pushes each change to a single subscriber pid as
{:arcadic_change, %Arcadic.Changes.Event{}} (see that struct for the marker
contract).
Reliability — best-effort at-most-once
ArcadeDB /ws has no replay or checkpoint. On any reconnect the process
delivers a :reconnected marker (before re-subscribing); on buffer overflow it
delivers an :overflow marker. Any marker obligates the consumer to
reconcile the affected database — the feed is a change hint, not a durable
log.
The subscriber may also receive an out-of-band error message
{:arcadic_change_error, reason}: :unauthorized (a terminal 401/403 on the
handshake — the process then stops, no reconnect spin) or :subscribe_rejected
(the server rejected a subscribe/unsubscribe with an error frame; non-terminal —
the socket stays open). Both reasons are bare atoms — the server's error text is
never forwarded.
Subscriber model
One subscriber pid per process. The first subscribe/3 binds the subscriber and
monitors it; later subscribes on other databases route to that same pid (demux
on Event.database). A subscribe with a different :subscriber is rejected
{:error, :subscriber_conflict}. The subscriber's :DOWN stops the process.
Backpressure
A bounded internal buffer (:max_buffer, default 1000) drops the oldest
events on overflow, emits an :overflow marker plus a
[:arcadic, :changes, :dropped] telemetry count per drain cycle in which events
were dropped (a short burst coalesces into a single marker; a sustained overflow
emits one roughly per drain cycle), and keeps reading the socket. The buffer
bounds arcadic's memory and guarantees a slow subscriber never wedges the
server (the socket is drained continuously); it does not bound the subscriber's
own mailbox — a persistently slow consumer must apply its own backpressure (or
reconcile on the markers and discard).
Absent optional dependency
The WebSocket client rides the optional mint_web_socket dependency. This
module always compiles; if the dependency is absent, start_link/1 returns
{:error, :mint_web_socket_not_available} at runtime. start_link/1 also
rejects a malformed :conn value-free — see Arcadic.Error for the
:invalid_auth / :invalid_url_scheme / :invalid_max_buffer reasons.
Telemetry
[:arcadic, :changes, :start | :stop | :disconnect | :dropped], all with
value-free metadata %{operation: :changes} (no database, record, or values).
Summary
Functions
Returns a specification to start this module under a supervisor.
Start the change-events process.
Subscribe client to change events on database.
Unsubscribe client from change events on database.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec start_link(keyword()) :: GenServer.on_start() | {:error, :mint_web_socket_not_available | :invalid_conn | :invalid_auth | :invalid_url_scheme | :invalid_max_buffer | :invalid_establish_timeout}
Start the change-events process.
Options
:conn— anArcadic.Conn(required). Supplies the/wsendpoint (base_url) and handshake credential (auth).:name— optionalGenServername.:max_buffer— bounded delivery buffer (default1000).
Returns {:error, :mint_web_socket_not_available} when the optional
mint_web_socket dependency is absent.
@spec subscribe(GenServer.server(), String.t(), keyword()) :: :ok | {:error, :subscriber_conflict}
Subscribe client to change events on database.
Options
:type— restrict to a document/vertex/edge type (default: all types).:change_types— a subset of[:create, :update, :delete](default: all).:subscriber— the pid to deliver events to. Defaults toself()evaluated here in the caller's process (never the process running the GenServer), so a supervisedChangesstill routes to the subscribing caller.
The first subscribe binds the subscriber; a later subscribe with a different
subscriber returns {:error, :subscriber_conflict}.
@spec unsubscribe(GenServer.server(), String.t()) :: :ok
Unsubscribe client from change events on database.