The loop over a set of feeds, as a value (spec/08 §11.4, D-140).
A monitor subscribes where a feed can push, polls where it cannot, and hands a chart
one finished delivery per slot. It is not a process: the process that owns the feeds —
a board's (#540), a LiveView's — holds it in its state, calls tick/2 each frame and
handle_message/2 from handle_info/2, and passes deliveries/1 and cache/1 to
Visualize.Chart.apply/2 or a compiled chart. That process subscribes, so the pushes
arrive in its mailbox.
Pushes buffer, and the next tick drains them: one merge, one trim and one reduction per
feed per tick, whatever the push rate. A pull feed is polled every every:
milliseconds. Errors and staleness are states, status/1, never raises.
Examples
iex> alias Visualize.Feed.Monitor
iex> signal = Visualize.Signals.new(:sine, [:t, :v], window: 100)
iex> feeds = %{wave: Visualize.Feed.Signals.new(signal, tick: 500)}
iex> monitor = Monitor.new(feeds, budget: 20) |> Monitor.tick(now: 0)
iex> Monitor.status(monitor)
%{wave: :live}
iex> %{wave: delivery} = Monitor.deliveries(monitor)
iex> {length(delivery.rows), delivery.reduced_by}
{20, :visualize}Summary
Functions
The monitor with a new budget, refinished at the next tick.
The schema cache, holding every version the deliveries name: apply/2's schemas:.
The finished delivery of every feed that has one, by name: a pool.
Each feed's last unresolved delivery reason, for the feeds that have one.
A message the owner received: {:ok, monitor} with a push buffered or a pushed schema
cached (spec/08 §11.3), :ignore for any other message. A push of a subscription the
monitor no longer holds is dropped.
A monitor over a map of feeds by slot name, calling nothing. Options: window: and
budget: (the request), reduce: (:lttb or :m4), every: (a pull feed's poll
interval, milliseconds, 1000), stale_after: (milliseconds without an answer before
a feed is :stale, nil for never) and limit: (the most rows held for a feed whose
rows cannot be trimmed by a window field, 100_000).
A feed added or replaced under a name: the old one unsubscribed, the new one described at the next tick.
Every feed fetched again at the next tick, push feeds included, whatever it holds (spec/08 §11.4): what a board does when a person sets its window (spec/15 §8.2).
A feed dropped, unsubscribed.
Each feed's state: :loading, :live, :stale or {:down, reason}.
Every push feed unsubscribed; the monitor's end.
One turn of the loop at now: (the owner's monotonic clock in milliseconds; read when
absent): every feed, in name order, is described if it is not yet, fetched if it holds
nothing, its window moved or its poll is due, its buffered pushes drained, every
version resolved, and its holding finished for the window and the budget.
The monitor with a new window. Each feed is asked the window that applies to its window
field (spec/08 §11.2, D-154) — a feed over ticks is asked nil, its own span, by a
window of instants, and is neither trimmed nor refetched as that window moves. A pull
feed fetches at the next tick; a push feed only when the window starts before what it
holds covers, and is otherwise trimmed.
Types
@type name() :: term()
@type status() :: :loading | :live | :stale | {:down, term()}
@type t() :: %Visualize.Feed.Monitor{ budget: Visualize.Feed.budget(), cache: Visualize.Data.Schema.Cache.t(), entries: %{required(name()) => map()}, every: non_neg_integer(), limit: pos_integer(), reduce: :lttb | :m4, refs: %{required(reference()) => name()}, stale_after: pos_integer() | nil, window: Visualize.Feed.window() }
Functions
@spec budget(t(), Visualize.Feed.budget()) :: t()
The monitor with a new budget, refinished at the next tick.
@spec cache(t()) :: Visualize.Data.Schema.Cache.t()
The schema cache, holding every version the deliveries name: apply/2's schemas:.
@spec deliveries(t()) :: %{required(name()) => Visualize.Data.Delivery.t()}
The finished delivery of every feed that has one, by name: a pool.
Each feed's last unresolved delivery reason, for the feeds that have one.
A message the owner received: {:ok, monitor} with a push buffered or a pushed schema
cached (spec/08 §11.3), :ignore for any other message. A push of a subscription the
monitor no longer holds is dropped.
@spec new(%{required(name()) => Visualize.Feed.t()}, keyword()) :: t()
A monitor over a map of feeds by slot name, calling nothing. Options: window: and
budget: (the request), reduce: (:lttb or :m4), every: (a pull feed's poll
interval, milliseconds, 1000), stale_after: (milliseconds without an answer before
a feed is :stale, nil for never) and limit: (the most rows held for a feed whose
rows cannot be trimmed by a window field, 100_000).
@spec put(t(), name(), Visualize.Feed.t()) :: t()
A feed added or replaced under a name: the old one unsubscribed, the new one described at the next tick.
Every feed fetched again at the next tick, push feeds included, whatever it holds (spec/08 §11.4): what a board does when a person sets its window (spec/15 §8.2).
A feed dropped, unsubscribed.
Each feed's state: :loading, :live, :stale or {:down, reason}.
@spec stop(t()) :: :ok
Every push feed unsubscribed; the monitor's end.
One turn of the loop at now: (the owner's monotonic clock in milliseconds; read when
absent): every feed, in name order, is described if it is not yet, fetched if it holds
nothing, its window moved or its poll is due, its buffered pushes drained, every
version resolved, and its holding finished for the window and the budget.
@spec window(t(), Visualize.Feed.window()) :: t()
The monitor with a new window. Each feed is asked the window that applies to its window
field (spec/08 §11.2, D-154) — a feed over ticks is asked nil, its own span, by a
window of instants, and is neither trimmed nor refetched as that window moves. A pull
feed fetches at the next tick; a push feed only when the window starts before what it
holds covers, and is otherwise trimmed.