Visualize.Feed.Monitor (Visualize v0.8.1)

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

Types

budgets()

One budget asked of every feed, or each feed's own by name, a name absent nil.

name()

status()

t()

Functions

budget(monitor, budget)

The monitor with a new budget, one or by feed name as new/2 takes it (spec/08 §11.4): a feed whose budget changed and that honors :budget is fetched again at the next tick, since it can serve a new budget only if it is asked; every other feed is finished again for its new budget from what it holds.

cache(monitor)

The schema cache, holding every version the deliveries name: apply/2's schemas:.

deliveries(monitor)

The finished delivery of every feed that has one, by name: a pool.

errors(monitor)

Each feed's last unresolved delivery reason, for the feeds that have one.

handle_message(monitor, arg2)

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.

new(feeds, opts \\ [])

A monitor over a map of feeds by slot name, calling nothing. Options: window: and budget: (the request; the budget is one for every feed, or a map from a feed's name to its own, a name it lacks asked nil), 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).

put(monitor, name, feed)

A feed added or replaced under a name: the old one unsubscribed, the new one described at the next tick.

refetch(monitor)

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).

remove(monitor, name)

A feed dropped, unsubscribed.

status(monitor)

Each feed's state: :loading, :live, :stale or {:down, reason}.

stop(monitor)

Every push feed unsubscribed; the monitor's end.

tick(monitor, opts \\ [])

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.

window(monitor, window)

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.