Filo.Stream (Filo v0.2.0)

Copy Markdown View Source

A single Hrana-over-HTTP stream: one GenServer owning one executor connection and the stream's baton sequence number.

In libsql's server a stream is a passive struct guarded by a mutex; each HTTP request acquires it, runs its pipeline, then releases it with a fresh baton. Filo collapses that into a process: the connection lives in the GenServer, and the mailbox serializes requests for free.

Sequence numbers

Every request must present the seq the server last handed out. A successful run advances seq by one, so each baton is single-use and requests on a stream are strictly serial. A request that presents the wrong seq is rejected as {:error, :baton_reused} without touching the connection.

Lifecycle

  • start_link/1 opens the connection through the executor; if the executor cannot open one, the start fails with {:open_failed, error}.
  • A close request releases the connection (via Filo.Request) and the process stops normally.
  • After :idle_timeout of inactivity the stream expires: it closes its connection and stops. Every successful run resets the timer.

The process is :temporary — a dead stream is not restarted; the client must open a new one.

Summary

Functions

run(server, seq, requests, opts \\ [])

Runs a pipeline of decoded Hrana requests against the stream.

run_cursor(server, seq, batch)

Runs a decoded Hrana batch map against the stream for a cursor, returning the raw Filo.BatchResult (which the caller streams as cursor entries) and the rotated sequence. Like run/3, a mismatched seq is {:error, :baton_reused}. The stream stays open.

start_link(opts)

Starts a stream.

Functions

run(server, seq, requests, opts \\ [])

@spec run(GenServer.server(), non_neg_integer(), [map()], keyword()) ::
  {:ok, Filo.Request.status(), [map()], non_neg_integer() | nil}
  | {:error, :baton_reused}

Runs a pipeline of decoded Hrana requests against the stream.

seq must match the sequence the stream currently expects. On success returns {:ok, status, results, next_seq} where status is :open (the stream continues, next_seq is the rotated sequence) or :closed (a close request released the connection, next_seq is nil). A mismatched seq returns {:error, :baton_reused}.

opts pass through to Filo.Request.handle/4's result encoders (rows: :json for the JSON transports' pre-encoded row fragments; default :maps for protobuf).

run_cursor(server, seq, batch)

@spec run_cursor(GenServer.server(), non_neg_integer(), map()) ::
  {:ok, Filo.BatchResult.t(), non_neg_integer()} | {:error, :baton_reused}

Runs a decoded Hrana batch map against the stream for a cursor, returning the raw Filo.BatchResult (which the caller streams as cursor entries) and the rotated sequence. Like run/3, a mismatched seq is {:error, :baton_reused}. The stream stays open.

start_link(opts)

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

Starts a stream.

Options:

  • :executor (required) — the Filo.Executor module.
  • :open_arg — host context passed to executor.open (default nil).
  • :open_context — the :authorize context threaded to executor.open/2 (default nil). See Filo.Executor.open/2.
  • :seq (required) — the initial sequence number.
  • :idle_timeout — inactivity timeout in ms (default 10000).
  • :name — a standard GenServer name (e.g. a Registry via-tuple).