# Tick streams

> Coalesce a feed's messages per key and apply one bounded batch a frame with createTickStream, so a burst never costs more than the screen can show.

A market feed sends far more messages than a screen can show. Writing each one straight into rows means thousands of updates to paint 60 frames a second, and the tab stalls. `createTickStream` from `bo-grid/realtime` sits between the socket and the grid. Pushing a message into it is cheap: no reactive write, no render. Once a frame it hands the grid the latest patch for each key that changed, at most `cap` of them.

Pick a feed rate below and compare the messages that come in with the patches the grid applies. At 5,000 messages a second over 12 rows, almost every message is merged into a patch still waiting for its frame.

Example: Coalescing a feed (tick-stream.svelte)

```svelte
<script lang="ts">
  import { Grid, type ColumnDef } from 'bo-grid';
  import { applyPatches, createRowIndex, createTickStream } from 'bo-grid/realtime';
  import { quotes } from './data';
  import { simulateFeed, type Tick } from './feed';

  // $state rows, written by applyPatches: only the fields that changed, and
  // only the latest tick for each row, once a frame.
  const rows = $state(quotes());
  const index = createRowIndex(rows, (r) => r.id);
  const columns: ColumnDef[] = [
    { type: 'text', key: 'symbol', header: 'Symbol', width: 100 },
    { type: 'price', key: 'price', header: 'Price', flash: 'auto' },
    { type: 'percent', key: 'change', header: 'Chg %' },
    { type: 'volume', key: 'volume', header: 'Volume', flex: 1 },
  ];

  let rate = $state(1000);
  let received = 0;
  let stats = $state({ received: 0, applied: 0, pending: 0 });
  const stream = createTickStream<number, Tick>({ apply: (batch) => applyPatches(index, batch) });

  $effect(() => {
    stream.start();
    // The simulated feed walks its own copy of the quotes (reading `rows`
    // here would make this effect re-run on every write).
    const stop = simulateFeed(quotes(), (id, tick) => {
      received++;
      stream.push(id, tick); // cheap: merges into the row's pending patch
    }, () => rate); // read on each beat: a new rate keeps the same feed
    const show = setInterval(() => (stats = { received, applied: stream.applied, pending: stream.pending }), 250);
    return () => {
      stop();
      clearInterval(show);
      stream.stop();
    };
  });
</script>

<div class="bar">
  <label>Feed <select bind:value={rate}><option value={100}>100 / s</option><option value={1000}>1,000 / s</option><option value={5000}>5,000 / s</option></select></label>
  <span>{stats.received.toLocaleString()} messages in</span>
  <span>{stats.applied.toLocaleString()} patches applied</span>
  <span>{stats.pending} waiting</span>
</div>
<Grid {rows} {columns} height={280} />

<style>
  .bar {
    display: flex;
    flex-wrap: wrap;
    align-items: center;
    gap: 6px 16px;
    margin: 0 0 8px;
    font-size: 12px;
  }
  select {
    font: inherit;
  }
</style>
```

data.ts:

```ts
// Small, fixed data for the docs' examples: every example reads the same
// symbols, so a reader compares features, not datasets.
// A type, not an interface: it then fits GridRow's index signature.
export type Quote = {
  id: number;
  symbol: string;
  name: string;
  sector: string;
  price: number;
  change: number;
  volume: number;
};

const BASE: Omit<Quote, 'id'>[] = [
  { symbol: 'VNM', name: 'Vinamilk', sector: 'Consumer', price: 68.5, change: 1.2, volume: 1_204_000 },
  { symbol: 'FPT', name: 'FPT Corp', sector: 'Technology', price: 132, change: -0.4, volume: 860_500 },
  { symbol: 'HPG', name: 'Hoa Phat', sector: 'Materials', price: 27.35, change: 0.74, volume: 3_412_900 },
  { symbol: 'VCB', name: 'Vietcombank', sector: 'Financials', price: 91.5, change: 0.55, volume: 1_986_200 },
  { symbol: 'MWG', name: 'Mobile World', sector: 'Consumer', price: 51.4, change: -1.1, volume: 2_250_400 },
  { symbol: 'SSI', name: 'SSI Securities', sector: 'Financials', price: 33.3, change: -0.45, volume: 4_104_700 },
  { symbol: 'VIC', name: 'Vingroup', sector: 'Real estate', price: 44.15, change: -0.11, volume: 1_530_100 },
  { symbol: 'GAS', name: 'PV Gas', sector: 'Energy', price: 71.8, change: 0.3, volume: 645_800 },
  { symbol: 'MSN', name: 'Masan', sector: 'Consumer', price: 76.2, change: 2.05, volume: 1_120_300 },
  { symbol: 'TCB', name: 'Techcombank', sector: 'Financials', price: 23.9, change: 0.84, volume: 5_870_000 },
  { symbol: 'VHM', name: 'Vinhomes', sector: 'Real estate', price: 41.6, change: -0.72, volume: 2_904_600 },
  { symbol: 'POW', name: 'PV Power', sector: 'Energy', price: 12.25, change: 1.66, volume: 6_315_200 },
];

/** A fresh copy of the quotes, so one example's edits never reach another. */
export function quotes(): Quote[] {
  return BASE.map((q, i) => ({ id: i + 1, ...q }));
}
```

feed.ts:

```ts
// A simulated market feed for the docs' examples: a random walk on each
// quote's price. In an app, this is your socket's message handler.
import type { Quote } from './data';

export type Tick = { price: number; change: number; volume: number };

/** Calls `onTick` about `perSecond` times a second (a number, or a function
    read on every beat, so the rate can change while it runs), for a random
    row each time. Returns a function that stops the feed. */
export function simulateFeed(rows: readonly Quote[], onTick: (id: number, tick: Tick) => void, perSecond: number | (() => number) = 20): () => void {
  const rate = typeof perSecond === 'function' ? perSecond : () => perSecond;
  const state = rows.map((r) => ({ id: r.id, ref: r.price / (1 + r.change / 100), price: r.price, volume: r.volume }));
  const every = 50;
  let due = 0;
  const timer = setInterval(() => {
    for (due += (rate() * every) / 1000; due >= 1; due--) {
      const s = state[Math.floor(Math.random() * state.length)];
      s.price = Math.max(0.01, Math.round(s.price * (1 + (Math.random() - 0.5) * 0.006) * 100) / 100);
      s.volume += Math.ceil(Math.random() * 40) * 100;
      onTick(s.id, { price: s.price, change: Math.round(((s.price - s.ref) / s.ref) * 10_000) / 100, volume: s.volume });
    }
  }, every);
  return () => clearInterval(timer);
}
```

## Creating a stream

With plain rows, apply each batch through the grid's `patchRows` ([Live updates](https://bonguynvan.github.io/bo-grid/docs/realtime/) explains why it is the fast path):

```ts
import { createTickStream } from 'bo-grid/realtime';

const stream = createTickStream<string, Partial<Quote>>({
  apply: (batch) => api.patchRows(batch), // once a frame, at most `cap` patches
});
stream.start();

socket.onmessage = (e) => {
  const q = JSON.parse(e.data);
  stream.push(q.symbol, q); // merges into the symbol's waiting patch
};
```

With `$state` rows, index them once and write the batch with `applyPatches`, which skips fields whose value did not change, so a chatty feed that re-sends the same price neither re-renders nor flashes:

```ts
import { applyPatches, createRowIndex, createTickStream } from 'bo-grid/realtime';

const index = createRowIndex(rows, (r) => r.symbol); // rebuild when symbols are added or removed
const stream = createTickStream<string, Partial<Quote>>({
  apply: (batch) => applyPatches(index, batch),
});
```

The key you push under is the feed's key. The grid's flash and selection follow its own row id (`getRowId`, `row.id` by default): keep that stable across updates.

## Coalescing

For each key, only the latest state is applied. Two patches that arrive for the same key between frames merge field by field when both are plain objects, so partial messages compose: `{ bid }` then `{ ask }` becomes `{ bid, ask }`. Anything else is last-write-wins. For updates that accumulate, pass `merge`:

```ts
const stream = createTickStream<string, { volume: number }>({
  apply: (batch) => api.patchRows(batch),
  merge: (prev, next) => ({ volume: prev.volume + next.volume }), // sum, don't replace
});
```

Keys drain in the order they first arrived, so a hot symbol cannot starve a quiet one: a key pushed again keeps its place in the queue.

A price board wants the latest state. A trade tape is the opposite case, where every trade must show: append trades to a `TradeTape` instead (see [Depth ladder](https://bonguynvan.github.io/bo-grid/docs/depth-ladder/#time-and-sales)).

## Feed health

- **`cap`** bounds the patches applied in one frame (400 by default). A burst cannot blow the frame budget. A sustained overload shows as a climbing `stream.pending` instead of dropped frames.
- **`pending`** counts the keys waiting for a frame: your feed-health number.
- **`applied`** counts the patches applied since the stream was created, a throughput counter. A batch whose `apply` threw is not counted.
- **`stop()`** pauses draining and keeps what waits, so a paused stream resumes with the latest state; `stop({ discard: true })` drops it. **`flushNow()`** applies everything at once, ignoring the cap: for a tab regaining focus, or a last paint before teardown.

If `apply` throws, that frame's batch is lost: it left the buffer before `apply` ran, and re-queuing it could apply it twice. The frame loop keeps going. Pass `onError` to handle the error; without it, it is re-thrown asynchronously so it does not vanish.

## Scheduling and tests

The stream drains on `requestAnimationFrame`, falling back to a 16 ms timer where there is none (server rendering, tests). Pass `schedule` to drive the frames yourself; that is also how the stream is unit-tested without a browser:

```ts
let frame: (() => void) | undefined;
const stream = createTickStream({ apply, schedule: (fn) => ((frame = fn), () => (frame = undefined)) });
stream.start();
stream.push('VNM', { price: 68.6 });
frame?.(); // one frame: apply ran with [['VNM', { price: 68.6 }]]
```

`bo-grid/realtime` is a separate entry with its own bundle, and none of it depends on the grid: importing it adds nothing to the grid core.

The live blotter below applies 200 fills a frame into 100,000 orders through a stream and `patchRows`:

Live example: https://bonguynvan.github.io/bo-grid/examples/liveblotter/ (source: https://github.com/bonguynvan/bo-grid/blob/main/src/demo/examples/)

## Reference

`createTickStream<K, P>(options)` returns a `TickStream<K, P>`.

| Option | Type | Default | What it does |
| --- | --- | --- | --- |
| `apply` | `(batch: Array<[K, P]>) => void` | required | Applies a frame's patches. Called at most once a frame, never with an empty batch. |
| `cap` | `number` | `400` | Most patches applied in one frame. |
| `merge` | `(prev: P, next: P) => P` | field-wise for plain objects | Combines two patches for one key between frames. |
| `schedule` | `(fn) => cancel` | `requestAnimationFrame` | The frame scheduler. |
| `onError` | `(err: unknown) => void` | re-thrown asynchronously | Hears an `apply` that threw. |

| Member | What it is |
| --- | --- |
| `push(key, patch)` | Takes one message's patch. |
| `start()` | Begins draining every frame. Idempotent. |
| `stop(opts?)` | Stops draining; `{ discard: true }` drops the waiting patches. |
| `flushNow()` | Applies everything waiting, ignoring the cap. |
| `pending` | Keys waiting for a frame. |
| `applied` | Patches applied since creation. |
| `running` | Whether it is draining. |

Helpers: `createRowIndex(rows, getRowId)` maps ids to rows; `applyPatches(index, batch)` writes a batch into indexed rows, skipping unchanged fields, and returns how many found a row. `TickBuffer` is the coalescing buffer on its own, for a loop you run yourself.

---
Page: https://bonguynvan.github.io/bo-grid/docs/tick-stream/ · All pages: https://bonguynvan.github.io/bo-grid/llms.txt