developer guide
WebSocket Feed for a JavaScript Data Grid
A socket delivers a snapshot and then a stream of deltas, and the data router takes both: router.load for the snapshot, router.apply for each delta, with a sequence number settling order and duplicates. One feed then drives every grid, chart and tile watching it, and a reconnect resumes rather than starting over.
Developer guide › Sources and pushdown › WebSocket Feed for a JavaScript Data Grid
Data Router: a live WebSocket feed
createDataRouter is built for a continuous live feed - ordering,
de-duplication, batching, resume after a drop, and moving a row between routes rather
than duplicating it are all shipped. What it does not do is open the connection.
The host owns the connection; the router owns everything after the message
arrives. That is a boundary worth stating plainly, because nothing in the
router's own name says so, and the two questions every integrator asks first - “does it handle the socket?” and “how do I send it data?” - both turn on it.
Wire a socket's messages to the router's two entry points
const router = createDataRouter({ rowKey: 'id', seq: 'v' });
router.attach(grid, () => true);
const socket = new WebSocket('wss://example.com/feed');
socket.onmessage = (event) => {
const msg = JSON.parse(event.data);
if (msg.kind === 'snapshot') router.load(msg.rows); // full keyed diff
else if (msg.kind === 'delta') router.apply(msg.changes); // in-place add/update/remove
};Two message kinds, two entry points. A snapshot
(the whole current world, sent once on connect or on resume) goes to
router.load(rows) - a plain array of records; it re-partitions
everything and applies a keyed diff per route, so rows that did not change do
not repaint. A delta (an incremental change) goes to
router.apply(deltas) - an array of { op: 'upsert'|'delete',
row, seq? } - which adds, updates or removes in place by rowKey,
moving a row between routes if its partition changed rather than duplicating it. Getting
this backwards is the mistake this guide exists to prevent: load() on every
message re-partitions the whole world on every tick (correct only for a snapshot);
apply() on the opening snapshot never seeds the store, so every route starts
empty. There is no third method for “a WebSocket message” - the host
reads kind (or whatever field its own wire format uses) and picks one of
these two.
No transport lock-in, and that is a benefit, not a gap. The router
takes rows, never a URL and never a socket object, so a real WebSocket, an
EventSource, a change-data-capture feed, a long-poll loop or an existing
message-bus client all wire up the same way - whatever arrives, hand the router the
snapshot array or the delta array. Nothing in packages/modules/data-router/
constructs a socket, so nothing there needs to change when the transport does.
Ordering, dedupe, and batching a fast feed
A socket is not a clean pipe: messages can arrive reordered, a reconnect can replay something already applied, and a fast feed can out-pace how often a grid should repaint. None of that is handled unless it is configured.
Without seq: a reordered packet wins silently - run and see it happen
const { createDataRouter } = await import('../packages/modules/data-router/index.js');
const rows = new Map();
const sink = { rows: { apply({ add = [], update = [] }) { for (const r of [...add, ...update]) rows.set(r.id, r); } } };
const router = createDataRouter({ rowKey: 'id' }); // no seq configured
router.attach(sink, () => true);
router.load([]);
router.apply([{ op: 'upsert', row: { id: 'x', price: 101 } }]); // the newer event
router.apply([{ op: 'upsert', row: { id: 'x', price: 100 } }]); // arrives late, but wins
return rows.get('x').price; // 100 - the stale packet clobbered the fresh oneWith seq (dedupe defaults on): the identical reorder, corrected
const { createDataRouter } = await import('../packages/modules/data-router/index.js');
const rows = new Map();
const sink = { rows: { apply({ add = [], update = [] }) { for (const r of [...add, ...update]) rows.set(r.id, r); } } };
const router = createDataRouter({ rowKey: 'id', seq: 'v' });
router.attach(sink, () => true);
router.load([]);
router.apply([{ op: 'upsert', row: { id: 'x', price: 101, v: 2 } }]);
router.apply([{ op: 'upsert', row: { id: 'x', price: 100, v: 1 } }]); // v:1 ≤ last seen v:2 - dropped
return `${rows.get('x').price} dropped=${router.dropped}`; // 101 dropped=1 - corrected, and countedWithout a seq, two rules to know. Within one
apply() call, last-writer-wins per rowKey - the last
element of the array for a given key is what lands, regardless of which one is
“newer” in real time. Across separate calls (separate socket messages), the
router has no way to tell a late, stale packet from a fresh one, so it applies whatever
arrives, whenever it arrives - an out-of-order delta lands out of order.
Configuring seq (a field name or fn(row)) fixes both: within a
batch the router sorts by seq before applying, and across calls it keeps a running
seqSeen per record and drops (into router.dropped) anything not
newer than what it already applied - which is exactly the reconnect/resume gate
below, working the same way for ordinary live reordering.
A fast feed: push plus batch/coalesce.
Call router.push(delta) instead of apply() and, with
coalesce: true or a batch: { intervalMs } configured, deltas
buffer instead of applying immediately; rapid updates to the same key settle to a single
apply on the timer (or on an explicit flushStream(), useful for a
deterministic point such as an animation frame). With no batching mode configured,
push applies at once, so it is always safe to feed a socket through
push rather than choosing between it and apply up front.
Reconnect and resume
Capture a cursor before the drop; replay after
// While live:
socket.onclose = () => {
const resumeFrom = router.lastSeq(); // ask the server to resume from here
const mark = router.checkpoint(); // or persist this per-record map instead
reconnect(resumeFrom);
};
// On the new connection, the server sends a fresh snapshot plus a replay
// that may include deltas already applied - the same onmessage handles it:
socket = new WebSocket(`wss://example.com/feed?since=${resumeFrom}`);
socket.onmessage = (event) => {
const msg = JSON.parse(event.data);
if (msg.kind === 'snapshot') router.load(msg.rows);
else if (msg.kind === 'delta') router.apply(msg.changes); // replays are dropped, not reapplied
};The pattern is snapshot-plus-replay, and the dedupe gate does the rest.
On reconnect, load a fresh snapshot (a keyed diff, so unchanged rows do not repaint) and
let the feed replay from around the last known point; any delta the router already
applied - because its seq is not newer than what checkpoint() holds for
that record - is dropped, so only genuinely new deltas advance the state.
seenThrough(mark) primes that same checkpoint from a persisted map (e.g. after
a page reload), so an early replay is dropped even before the router has applied anything
itself in this session. None of this requires the socket to be gone - the router has
no idea whether it is talking to the first connection or the fifth.
The knobs a live feed makes you reach for
Four factory options and one route option matter once the feed is real rather than a
fixture, and none of them is needed to get started. onUnrouted is the sink
for whatever matched no route - it receives the row on load and
query, and the whole delta on apply - and is the honest
alternative to an attachDefault grid when a stray record is a bug to log
rather than a row to show. selectionDebounce (ms, default 16) is how long a
linked grid waits before refiltering on a selection change; set it to 0 in a
test so every selection.set refilters before the next statement.
metricsInterval (ms, default 1000) is the cadence of the
on('metrics') emit, which runs only while a listener is registered; a
devtools panel inherits it. On a route, backpressure ({ maxHz,
minInterval, sample, maxLag }) throttles how often that one viewer repaints under
load without touching what it holds - a chart that cannot draw 200 times a second
gets { maxHz: 10 }, and the grid beside it stays live.
Persistence survives a reload: choose the store once
router.persist({ key: 'orders', dbName: 'lattice', storeName: 'router' });
await router.restore(); // true when a snapshot was found and applied
persist writes the router's state to IndexedDB under dbName /
storeName (defaults are provided; pass your own indexedDB to
test against a fake, or a storage pair of get/set
to use something else entirely). persisting reads false when the
store was unavailable and the router degraded to memory, which is the case to check
before promising a user their view will be there tomorrow.
Several independent feeds, one router
Two sockets, two sources, one merged store
const orders = router.addSource('orders');
const inventory = router.addSource('inventory');
const ordersSocket = new WebSocket('wss://example.com/orders');
ordersSocket.onmessage = (event) => {
const msg = JSON.parse(event.data);
if (msg.kind === 'snapshot') orders.load(msg.rows);
else if (msg.kind === 'delta') orders.apply(msg.changes);
};
// inventorySocket wired the same way, against the `inventory` handleaddSource(id) returns a per-feed handle - its own
load/apply/push/remove - so each
socket is wired to its own handle exactly as a single feed is wired to the router
directly; the router merges every source into one keyed store, namespaced by
key when two feeds' ids could otherwise collide. There is no separate lookup
method to re-fetch a handle later - keep the reference addSource
returns, the same way the code above keeps orders and
inventory.
No backend yet: MockWebSocket frames messages identically
import { MockWebSocket, opsFeed } from '@toclocoinc/lattice-grid/modules/mock-socket';
const socket = new MockWebSocket({ feed: opsFeed({ seed: 7 }) });
// everything above this line is the only thing that changes going live:
// const socket = new WebSocket('wss://example.com/feed');Every example on this page is executed, against the real module and
MockWebSocket, in demo/router-websocket.mjs
(node demo/router-websocket.mjs) - run it to see each property above as
output rather than prose.