Streaming data

layer.append(batch) streams new features into an existing layer without re-projecting the features already on screen — the key to live data like species occurrences. Points land in patchy, multi-scale clusters (a Thomas-like point process) and are clipped to land; the polygon “ranges” are large and rendered very transparently, so overlapping ranges build up species-richness hotspots. A live readout (top-right) shows the record count, percentage of the chosen total, and the average append speed so far.

Each example keeps its own retained GeoJSON array alongside the layer. Every feature starts the same color; Randomize colors picks a new color, applies it to all future features, and rewrites the retained features’ properties.color then recolor()s them — so streaming and the retained-for-redraw data stay in sync. Use Stream to pause/resume, Restart for a fresh stream, Data size for the total to stream (100k / 1M / 10M), Batch size for how many features each tick emits — including adaptive (the default), which auto-tunes the batch each frame to fit a ~30fps budget so the stream runs as fast as the backend allows while staying responsive — and Batch delay (ms) for an artificial per-batch delay that mirrors loading from a file. The readout’s batch line shows the live (adaptive) batch size.

The source (makeStreamingPoints / makeStreamingPolygons in shared/geo-data.ts) is an async generator that yields batches lazily up to the chosen total, so nothing is materialized up front.

Backend matters. The point examples default to the canvas backend (the polygons example defaults to auto), where append is truly O(new) (new geometry is drawn on top, no full redraw) — so adaptive batching ramps up and the stream stays fast. On WebGL, append currently rebuilds the whole layer renderer per batch (O(total)); adaptive then backs off the batch to hold the frame budget (you’ll see the batch readout fall toward 1) and throughput stays modest. A follow-up adds true O(new) WebGL sub-buffer uploads; until then, use canvas for high streaming throughput.

Streaming points on a world map

autowebglcanvassvg

Export PNG

fps 0frame 0 ms

Stream

runpause

RestartRandomize colors

draw.ts

draw.ts

import { geoEquirectangular } from "d3-geo";
import { geoMap, type LayerHandle } from "@mapequation/d3gl/map";
import {
  loadWorld,
  makeStreamingPoints,
  DEFAULT_STREAM_COLOR,
  type StreamPoint,
} from "../shared/geo-data.js";
import { StreamController, randomHsl, DATA_SIZE_TOTALS, ADAPTIVE_SEED_BATCH } from "../shared/streaming.js";
import { createStatsOverlay } from "../shared/stats-overlay.js";
import type { ImperativeSetup } from "../types.js";

const OCEAN = "#dbe7f3";
const LAND = "#e9e7df";

/**
 * Stream points (clustered around cities + rivers) and append them live with
 * `layer.append(batch)` — only the NEW points project; existing ones are
 * untouched and clipped to land. Every feature carries a `color` the example
 * owns: all start red, and "Randomize colors" picks a new color for FUTURE
 * points AND rewrites the retained ones' properties + `recolor()`s them — showing
 * streaming and retained-for-redraw working together.
 */
export const setup: ImperativeSetup = (host, { width, height, backend, options }) => {
  const world = loadWorld();
  const projection = geoEquirectangular().fitSize([width, height], { type: "Sphere" });
  const map = geoMap(host, { width, height, projection, backend });
  map.layer("ocean", [world.sphere], { fill: OCEAN });
  map.layer("land", [world.land], { fill: LAND, stroke: "#9aa3ad", lineWidth: 0.4 });
  map.enableZoom([1, 40]);

  const retained: StreamPoint[] = [];
  let currentColor = DEFAULT_STREAM_COLOR; // the color new points get (until randomized)
  const pointOpts = {
    fill: (f: StreamPoint) => f.properties.color,
    pointRadius: 0.5,
    id: (f: StreamPoint) => f.properties.id,
    clipTo: "land", // only show points over land
    pickable: false, // millions of streamed points — skip the hit index (no hover) to save memory
  };
  let points: LayerHandle<StreamPoint> = map.layer("points", [], pointOpts);
  map.render();

  const stats = createStatsOverlay(host);
  let count = 0; // records appended this session
  let appendMs = 0; // total time spent in append() — gives the average speed

  let seed = 1;
  let total = DATA_SIZE_TOTALS[String(options.size)] ?? 1_000_000;
  const ctrl = new StreamController<StreamPoint>({
    source: (o) => makeStreamingPoints({ total, ...o }),
    onBatch: (batch) => {
      for (const f of batch) {
        f.properties.color = currentColor; // new points get the current color
        retained.push(f); // retain for redraw / recolor (loop, not push(...spread): batch can be 1M)
      }
      const t0 = performance.now();
      points.append(batch); // incremental append — only the new points project
      appendMs += performance.now() - t0;
      count += batch.length;
      stats.update(count, total, count / (appendMs / 1000), ctrl.batchSize);
    },
    onReset: () => {
      retained.length = 0;
      count = 0;
      appendMs = 0;
      points = map.layer("points", [], pointOpts); // re-register empty to clear
      stats.update(0, total, 0, ctrl.batchSize, true);
    },
  });
  // "adaptive" auto-tunes the batch to a frame budget; a number fixes it.
  const applyBatch = (opt: string): void => {
    ctrl.adaptive = opt === "adaptive";
    ctrl.batchSize = ctrl.adaptive ? ADAPTIVE_SEED_BATCH : Number(opt);
  };
  // Stream runs only when the user wants it AND the canvas is on-screen (the harness
  // calls setVisible). A manual pause persists across scroll-out/in.
  let visible = true;
  let userRunning = options.stream === "run";
  let lastBatchOpt = String(options.batch);
  applyBatch(lastBatchOpt);
  ctrl.delayMs = Number(options.rate);
  ctrl.setRunning(userRunning && visible);
  ctrl.restart(seed);

  let lastRandomize = Number(options.randomize) || 0;
  let lastRestart = Number(options.restart) || 0;

  return {
    engine: map,
    render: (o) => {
      userRunning = o.stream === "run";
      ctrl.setRunning(userRunning && visible);
      if (o.randomize !== lastRandomize) {
        lastRandomize = Number(o.randomize) || 0;
        currentColor = randomHsl(); // future points...
        for (const f of retained) f.properties.color = currentColor; // ...and already-stored ones
        points.recolor(); // re-render from the retained, updated properties
      }
      const batchOpt = String(o.batch);
      const rate = Number(o.rate);
      const newTotal = DATA_SIZE_TOTALS[String(o.size)] ?? total;
      if (
        batchOpt !== lastBatchOpt ||
        rate !== ctrl.delayMs ||
        newTotal !== total ||
        o.restart !== lastRestart
      ) {
        lastRestart = Number(o.restart) || 0;
        lastBatchOpt = batchOpt;
        applyBatch(batchOpt);
        ctrl.delayMs = rate;
        total = newTotal;
        ctrl.restart(++seed);
      }
    },
    setVisible: (v) => {
      visible = v;
      ctrl.setRunning(userRunning && visible); // pause offscreen; resume only if user wants run
    },
    dispose: () => {
      ctrl.dispose();
      stats.destroy();
    },
  };
};

geo-data.ts

geo-data.ts

import { geoGraticule } from "d3-geo";
import type { Feature, FeatureCollection, LineString, MultiLineString, MultiPoint, MultiPolygon, Point, Polygon } from "geojson";
import { feature } from "topojson-client";
import land110m from "world-atlas/land-110m.json";

/** A synthetic grid cell with a continuous value and a categorical bioregion. */
export interface Cell {
  id: string;
  geometry: Polygon;
  /** Cell centroid [lon, lat]. */
  center: [number, number];
  /** Continuous field in [0, 1] (value field). */
  value: number;
  /** Categorical bioregion id in 0..7. */
  bioregion: number;
}

/** Base cell size in degrees; the example scales it by powers of two via a slider. */
export const BASE_STEP = 1;

function clamp01(x: number): number {
  return Math.max(0, Math.min(1, x));
}

/** Generate a global grid of `step`°×`step`° cells with smooth synthetic fields. */
export function makeCells(step: number = BASE_STEP): Cell[] {
  const cells: Cell[] = [];
  let col = 0;
  for (let lon = -180; lon < 180; lon += step, col++) {
    let row = 0;
    for (let lat = -90; lat < 90; lat += step, row++) {
      const lonR = (lon * Math.PI) / 180;
      const latR = (lat * Math.PI) / 180;
      const value = clamp01(0.5 + 0.5 * Math.sin(lonR * 2) * Math.cos(latR * 3));
      const field = (Math.sin(lon / 40) + Math.cos(lat / 30)) * 0.5 + 1; // ~[0,2]
      const bioregion = Math.min(7, Math.max(0, Math.floor((field / 2) * 8)));
      const geometry: Polygon = {
        type: "Polygon",
        coordinates: [
          [
            [lon, lat],
            [lon, lat + step],
            [lon + step, lat + step],
            [lon + step, lat],
            [lon, lat],
          ],
        ],
      };
      cells.push({
        id: `${col}-${row}`,
        geometry,
        center: [lon + step / 2, lat + step / 2],
        value,
        bioregion,
      });
    }
  }
  return cells;
}

/** A fine grid over the central third of the globe (lon ±60°, lat ±30°), 4° cells —
 *  a "dense" demo layer (used clipped to land). */
export function centreCells(): Cell[] {
  return makeCells(4).filter((c) => Math.abs(c.center[0]) <= 60 && Math.abs(c.center[1]) <= 30);
}

/** Wrap cells as a FeatureCollection for projection fitting. */
export function cellsToFeatureCollection(cells: readonly Cell[]): FeatureCollection {
  const features: Feature[] = cells.map((c) => ({
    type: "Feature",
    properties: { id: c.id },
    geometry: c.geometry,
  }));
  return { type: "FeatureCollection", features };
}

/** A GeoJSON object d3-geo can fill that isn't part of the strict GeoJSON spec. */
export type Sphere = { type: "Sphere" };

/** The land outline (Natural Earth 110m) plus a sphere to fill as ocean. */
export interface World {
  sphere: Sphere;
  land: MultiPolygon;
}

// Derive the topojson Topology type from feature()'s own signature so we don't
// take a direct dependency on the (transitive) topojson-specification types.
type Topology = Parameters<typeof feature>[0];

/**
 * Convert the bundled world-atlas TopoJSON into a land MultiPolygon and a sphere.
 */
export function loadWorld(): World {
  const topo = land110m as unknown as Topology;
  const fc = feature(topo, topo.objects.land!) as unknown as FeatureCollection<MultiPolygon>;
  return { sphere: { type: "Sphere" }, land: fc.features[0]!.geometry };
}

/** A few well-known cities to show point geometry rendered alongside the grid. */
export interface City {
  id: string;
  name: string;
  geometry: Point;
}

export function makeCities(): City[] {
  const places: [string, number, number][] = [
    ["London", -0.13, 51.51],
    ["New York", -74.01, 40.71],
    ["Tokyo", 139.69, 35.69],
    ["Sydney", 151.21, -33.87],
    ["Cape Town", 18.42, -33.92],
    ["Rio de Janeiro", -43.2, -22.91],
    ["Nairobi", 36.82, -1.29],
    ["Mumbai", 72.88, 19.08],
  ];
  return places.map(([name, lon, lat]) => ({
    id: name!,
    name: name!,
    geometry: { type: "Point", coordinates: [lon!, lat!] },
  }));
}

/** A 20° graticule as one MultiLineString feature. */
export function makeGraticule(): Feature<MultiLineString> {
  return { type: "Feature", properties: {}, geometry: geoGraticule().step([20, 20])() };
}

/** A great-circle-ish route as a LineString feature (London -> New York -> Tokyo). */
export function makeRoute(): Feature<LineString> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "LineString", coordinates: [[-0.13, 51.51], [-74.01, 40.71], [139.69, 35.69]] },
  };
}

/** A cluster of locations as one MultiPoint feature. */
export function makeCluster(): Feature<MultiPoint> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "MultiPoint", coordinates: [[18.42, -33.92], [151.21, -33.87], [-43.2, -22.91], [36.82, -1.29], [72.88, 19.08]] },
  };
}

/** A standalone Polygon feature (a box over the Sahara) to showcase polygon geometry.
 *  Wound CLOCKWISE so d3-geo fills the small box (not the sphere complement). */
export function makeDemoPolygon(): Feature<Polygon> {
  return {
    type: "Feature", properties: { name: "demo-region" },
    geometry: { type: "Polygon", coordinates: [[[0, 15], [0, 30], [30, 30], [30, 15], [0, 15]]] },
  };
}

/** A handful of major rivers as rough named polylines (the bundled world-atlas data
 *  has no rivers), shown on the GeoJSON-features map and used as streaming cluster
 *  centers. Coordinates are approximate [lon, lat] traces, mouth → source. */
export function makeMajorRivers(): Feature<LineString, { name: string }>[] {
  const rivers: [string, [number, number][]][] = [
    ["Amazon", [[-50.0, -0.7], [-55.5, -2.5], [-60.0, -3.1], [-67.9, -3.5], [-73.2, -4.5]]],
    ["Nile", [[31.3, 31.4], [32.9, 24.1], [32.5, 15.6], [32.5, 9.5], [31.6, 2.3]]],
    ["Mississippi", [[-89.2, 29.2], [-90.1, 32.3], [-90.2, 38.6], [-91.2, 43.5], [-95.0, 47.2]]],
    ["Yangtze", [[121.8, 31.4], [114.3, 30.6], [106.5, 29.6], [100.2, 26.9], [94.7, 33.5]]],
    ["Congo", [[12.4, -6.0], [16.2, -4.3], [20.0, -1.0], [25.2, 0.5], [27.2, 3.0]]],
    ["Volga", [[48.0, 46.3], [45.0, 48.7], [44.5, 51.6], [47.5, 54.3], [37.0, 57.3]]],
    ["Ganges", [[90.5, 22.5], [88.0, 24.5], [83.0, 25.4], [78.0, 26.5], [78.9, 30.1]]],
  ];
  return rivers.map(([name, coordinates]) => ({
    type: "Feature",
    properties: { name },
    geometry: { type: "LineString", coordinates },
  }));
}

// ---------------------------------------------------------------------------
// Streaming sources — async generators that emit batches of features lazily
// (only `batchSize` are materialized per tick, never the whole `total`), so a
// consumer can `await`-iterate and append them live. Points/cells are CLUSTERED
// around cities + major-river vertices (not uniform), which the world-map
// examples then clip to land. Used by the "streaming data" examples.
// ---------------------------------------------------------------------------

/** A solid default color all streamed features start with (the example's
 *  "randomize" button swaps in a new color for new + retained features). */
export const DEFAULT_STREAM_COLOR = "#e23b2f";

/** Per-feature properties: a stable id (continues across batches) + a color the
 *  example owns (constant by default; swapped by the randomize button). */
export interface StreamProps {
  id: number;
  color: string;
}
export type StreamPoint = Feature<Point, StreamProps>;
export type StreamPolygon = Feature<Polygon, StreamProps>;

export interface StreamOptions {
  /** Total features emitted before the generator completes. Default 10,000,000. */
  total?: number;
  /** Features per yielded batch. A function is re-read every batch, so the caller can
   *  resize adaptively (e.g. to fit a frame budget). Default 1000. */
  batchSize?: number | (() => number);
  /** Artificial delay between batches (ms), to mirror loading from a file/network.
   *  Even 0 yields a macrotask so the browser can paint between batches. Default 0. */
  delayMs?: number;
  /** Seed for the deterministic PRNG (reproducible streams). Default 1. */
  seed?: number;
  /** Cooperative cancellation: iteration stops once `signal.aborted` is true. */
  signal?: { aborted: boolean };
}

/** Small, fast, seedable PRNG (mulberry32) — deterministic so streams reproduce. */
function mulberry32(seed: number): () => number {
  let a = seed >>> 0;
  return () => {
    a |= 0;
    a = (a + 0x6d2b79f5) | 0;
    let t = Math.imul(a ^ (a >>> 15), 1 | a);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
  };
}

const clamp = (x: number, lo: number, hi: number): number => Math.max(lo, Math.min(hi, x));

/**
 * A clustered point process (Thomas-like): many weighted "hotspot" parents at a
 * range of scales — tight, dense cities; medium clumps along rivers; and a field
 * of random global hotspots with varied spread/weight. Offspring fall around a
 * parent with an exponential (heavy-tailed) radius, so the result is patchy and
 * multi-scale — far closer to real species-occurrence density than a smooth
 * gaussian. Clip-to-land then carves the continental outline.
 */
interface Parent {
  lon: number;
  lat: number;
  /** Mean offspring radius in degrees. */
  spread: number;
  /** Relative share of points drawn from this parent. */
  weight: number;
}

function buildParents(rng: () => number): Parent[] {
  const parents: Parent[] = [];
  for (const c of makeCities())
    parents.push({ lon: c.geometry.coordinates[0]!, lat: c.geometry.coordinates[1]!, spread: 1.5, weight: 3 });
  for (const river of makeMajorRivers())
    for (const p of river.geometry.coordinates) parents.push({ lon: p[0]!, lat: p[1]!, spread: 2.5, weight: 2 });
  // Random global hotspots: rng()*rng() biases toward small spreads/weights, so
  // most clumps are tight with a few diffuse ones — a wide range of scales.
  for (let i = 0; i < 240; i++) {
    parents.push({
      lon: rng() * 360 - 180,
      lat: rng() * 160 - 80,
      spread: 0.6 + 9 * rng() * rng(),
      weight: 0.2 + 3 * rng() * rng(),
    });
  }
  return parents;
}

function cumulativeWeights(parents: readonly Parent[]): number[] {
  const cum: number[] = [];
  let s = 0;
  for (const p of parents) {
    s += p.weight;
    cum.push(s);
  }
  return cum;
}

/** Pick a parent by weight (binary search over cumulative weights). */
function pickParent(rng: () => number, parents: readonly Parent[], cum: readonly number[]): Parent {
  const r = rng() * cum[cum.length - 1]!;
  let lo = 0;
  let hi = cum.length - 1;
  while (lo < hi) {
    const mid = (lo + hi) >> 1;
    if (cum[mid]! < r) lo = mid + 1;
    else hi = mid;
  }
  return parents[lo]!;
}

/** Offspring location around a parent: exponential radius, uniform direction. */
function clusteredLonLat(rng: () => number, p: Parent): [number, number] {
  const radius = -p.spread * Math.log(Math.max(1e-9, rng())); // exponential, mean = spread
  const ang = rng() * 2 * Math.PI;
  return [
    clamp(p.lon + radius * Math.cos(ang), -180, 180),
    clamp(p.lat + radius * Math.sin(ang), -90, 90),
  ];
}

/**
 * An irregular, star-convex polygon ring (3–10 vertices, varied per-vertex radius)
 * centered at [clon, clat] with overall extent ≤ ~`size`° — a rough species range.
 * Angles are evenly spaced with bounded jitter so they stay monotonic ⇒ the ring is
 * simple (non-self-intersecting) and closed. Longitude offsets are widened by
 * 1/cos(lat) so ranges don't look squished toward the poles.
 *
 * WINDING: built CLOCKWISE in [lon, lat] (note the NEGATIVE angle). d3-geo fills on
 * the sphere, and a small exterior ring wound counter-clockwise is treated as its
 * complement → it fills the whole map. See `AGENTS.md` and `geo/project.ts`.
 */
function randomRangeRing(rng: () => number, clon: number, clat: number, size: number): [number, number][] {
  const verts = 3 + Math.floor(rng() * 8); // 3..10
  // Strongly heavy-tailed size: the vast majority of ranges are TINY and only ~3% are
  // visibly large. At high counts the translucent fill then reads as a density gradient
  // (clustered richness hotspots) instead of saturating the whole map red.
  const base =
    rng() < 0.03
      ? size * (0.1 + 0.15 * rng()) // ~3% larger ranges: 0.10..0.25 * size
      : size * (0.02 + 0.07 * rng() * rng()); // most tiny: 0.02..0.09 * size, biased small (rng²)
  const latScale = 1 / Math.max(0.25, Math.cos((clat * Math.PI) / 180));
  const ring: [number, number][] = [];
  for (let i = 0; i < verts; i++) {
    const ang = -((i + 0.5 * rng()) / verts) * 2 * Math.PI; // NEGATIVE ⇒ clockwise ⇒ fills interior
    const r = base * (0.5 + rng()); // per-vertex radius variation
    ring.push([
      clamp(clon + r * Math.cos(ang) * latScale, -180, 180),
      clamp(clat + r * Math.sin(ang), -90, 90),
    ]);
  }
  ring.push(ring[0]!); // close the ring
  return ring;
}

const tick = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));

/** Stream world points as GeoJSON `Feature<Point>` batches, clustered around
 *  many multi-scale hotspots. All start with DEFAULT_STREAM_COLOR. */
export async function* makeStreamingPoints(opts: StreamOptions = {}): AsyncGenerator<StreamPoint[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPoint[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [lon, lat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Point", coordinates: [lon, lat] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

/** Stream irregular polygon "ranges" clustered around hotspots — each a 3–10
 *  vertex star-convex polygon of varied size ≤ ~`size`°, to mimic species ranges.
 *  All start with DEFAULT_STREAM_COLOR; the example renders them very transparent
 *  so overlapping ranges build up richness. */
export async function* makeStreamingPolygons(
  opts: StreamOptions & { size?: number } = {},
): AsyncGenerator<StreamPolygon[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, size = 16, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPolygon[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [clon, clat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Polygon", coordinates: [randomRangeRing(rng, clon, clat, size)] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

streaming.ts

streaming.ts

/**
 * Drives a streaming source (an async batch generator) into a consumer, with
 * run/pause and restart — the plumbing shared by the streaming examples so each
 * `draw.ts` only has to say HOW to append a batch (the d3gl-specific part).
 */

export interface StreamSourceOpts {
  /** Re-read every batch (the controller may resize it adaptively). */
  batchSize: () => number;
  delayMs: number;
  seed: number;
  signal: { aborted: boolean };
}

export interface StreamControllerOpts<T> {
  /** Build a fresh source generator for a (re)started session. */
  source: (o: StreamSourceOpts) => AsyncGenerator<T[]>;
  /** Consume one freshly-arrived batch (append it to the layer + retain it). */
  onBatch: (batch: T[]) => void;
  /** Clear retained state + the layer at the start of a (re)started session. */
  onReset: () => void;
}

/**
 * One streaming session at a time. `restart(seed)` resets and starts a new
 * session (superseding any running one); `setRunning(false)` pauses the pump;
 * `dispose()` stops everything for good.
 *
 * Adaptive mode (`adaptive = true`): after each batch the controller times how long
 * `onBatch` (build + append + paint) took and nudges `batchSize` toward the number
 * of features that fit `frameBudgetMs` — so the stream runs as fast as the backend
 * allows while staying responsive (the old Bioregions "fit ~30fps" trick). The source
 * re-reads `batchSize` every batch via the getter passed in `restart`.
 */
export class StreamController<T> {
  batchSize = 1000;
  delayMs = 0;
  /** When true, `batchSize` is auto-tuned each batch to fit `frameBudgetMs`. */
  adaptive = false;
  /** Per-batch time budget for adaptive mode (≈30fps). */
  frameBudgetMs = 32;
  private session = 0;
  private running = true;
  private disposed = false;

  constructor(private readonly opts: StreamControllerOpts<T>) {}

  setRunning(v: boolean): void {
    this.running = v;
  }

  /** Reset + start a fresh session (supersedes any in flight). */
  restart(seed: number): void {
    if (this.disposed) return;
    this.opts.onReset();
    void this.pump(seed);
  }

  /** Stop the current session permanently (call from the example's dispose). */
  dispose(): void {
    this.disposed = true;
    this.session++;
  }

  private async pump(seed: number): Promise<void> {
    const my = ++this.session;
    const signal = { aborted: false };
    // Getter so adaptive resizes take effect on the very next batch the source pulls.
    const gen = this.opts.source({ batchSize: () => this.batchSize, delayMs: this.delayMs, seed, signal });
    for await (const batch of gen) {
      // Superseded (restart) or disposed → abort this session.
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      // Paused → idle without consuming, until resumed / superseded / disposed.
      while (!this.running && !this.disposed && my === this.session) {
        await new Promise((r) => setTimeout(r, 60));
      }
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      const t0 = performance.now();
      this.opts.onBatch(batch);
      if (this.adaptive) {
        // Steer toward the budget: scale by budget/elapsed, clamped to ≤2× per step so
        // it ramps quickly but doesn't oscillate. Bounded to [1, 1_000_000].
        const dt = Math.max(0.5, performance.now() - t0);
        const factor = Math.max(0.5, Math.min(2, this.frameBudgetMs / dt));
        this.batchSize = Math.max(1, Math.min(1_000_000, Math.round(this.batchSize * factor)));
      }
    }
  }
}

/** A random vivid color (used by the "randomize colors" button). Pass `alpha` < 1
 *  for translucent fills (e.g. overlapping polygon ranges). */
export function randomHsl(alpha = 1): string {
  const h = Math.floor(Math.random() * 360);
  return alpha >= 1 ? `hsl(${h}, 70%, 55%)` : `hsla(${h}, 70%, 55%, ${alpha})`;
}

/** The batch-size choices offered by the streaming examples' control. "adaptive"
 *  auto-tunes the batch to fit a frame budget (see StreamController.adaptive). */
export const BATCH_SIZES = ["adaptive", "1", "10", "100", "1000", "100000", "1000000"];
/** Seed batch size when adaptive mode starts, before it converges. */
export const ADAPTIVE_SEED_BATCH = 256;
/** The artificial per-batch delay choices (ms) — mirrors loading from a file. */
export const RATES_MS = ["0", "16", "100", "500"];
/** Data-size choices (total features to stream) and their numeric totals. */
export const DATA_SIZES = ["100k", "1M", "10M"];
export const DATA_SIZE_TOTALS: Record<string, number> = {
  "100k": 100_000,
  "1M": 1_000_000,
  "10M": 10_000_000,
};

stats-overlay.ts

stats-overlay.ts

/**
 * A tiny HTML overlay (top-right of the canvas) showing how many records have
 * streamed in — count, percentage of the total, and the average append speed so
 * far (records / time spent in `append`). Mirrors the Infomap Bioregions readout.
 */
export interface StatsOverlay {
  /** Update the readout. Throttled to ~10 Hz unless `force`. `batch` (optional) shows
   *  the current batch size — handy to watch adaptive batching converge. */
  update(count: number, total: number, recordsPerSec: number, batch?: number, force?: boolean): void;
  destroy(): void;
}

export function createStatsOverlay(host: HTMLElement): StatsOverlay {
  const el = document.createElement("div");
  el.className =
    "absolute top-2 right-2 pointer-events-none rounded-md bg-white/80 px-2 py-1 " +
    "font-mono text-[11px] leading-tight text-[#333] shadow-sm [font-variant-numeric:tabular-nums]";
  host.appendChild(el);

  const fmt = (n: number): string => Math.round(n).toLocaleString();
  let last = 0;

  return {
    update(count, total, recordsPerSec, batch, force = false) {
      const now = performance.now();
      if (!force && now - last < 100) return;
      last = now;
      const ratio = total > 0 ? count / total : 0;
      const pct = (ratio * 100).toFixed(ratio < 0.1 ? 1 : 0);
      const batchLine = batch != null ? `<br>batch <b>${fmt(batch)}</b>` : "";
      el.innerHTML = `records <b>${fmt(count)}</b> (${pct}%)<br>speed <b>${fmt(recordsPerSec)}</b> rec/s${batchLine}`;
    },
    destroy() {
      el.remove();
    },
  };
}

Streaming polygons on a world map

autowebglcanvassvg

Export PNG

fps 0frame 0 ms

Stream

runpause

RestartRandomize colors

draw.ts

draw.ts

import { geoEquirectangular } from "d3-geo";
import { geoMap, type LayerHandle } from "@mapequation/d3gl/map";
import { loadWorld, makeStreamingPolygons, type StreamPolygon } from "../shared/geo-data.js";
import { StreamController, randomHsl, DATA_SIZE_TOTALS, ADAPTIVE_SEED_BATCH } from "../shared/streaming.js";
import { createStatsOverlay } from "../shared/stats-overlay.js";
import type { ImperativeSetup } from "../types.js";

const OCEAN = "#dbe7f3";
const LAND = "#e9e7df";
const RANGE_ALPHA = 0.05; // very transparent so overlapping ranges build up a density gradient
const DEFAULT_RANGE_COLOR = `hsla(8, 80%, 53%, ${RANGE_ALPHA})`; // translucent red

/**
 * Stream small polygon cells (clustered around cities + rivers) and append them
 * live with `layer.append`. Only new cells project/tessellate per batch; existing
 * ones stay put, clipped to land. All cells start red; "Randomize colors" sets the
 * color for future cells and rewrites + `recolor()`s the retained ones.
 */
export const setup: ImperativeSetup = (host, { width, height, backend, options }) => {
  const world = loadWorld();
  const projection = geoEquirectangular().fitSize([width, height], { type: "Sphere" });
  const map = geoMap(host, { width, height, projection, backend });
  map.layer("ocean", [world.sphere], { fill: OCEAN });
  map.layer("land", [world.land], { fill: LAND, stroke: "#9aa3ad", lineWidth: 0.4 });
  map.enableZoom([1, 40]);

  const retained: StreamPolygon[] = [];
  let currentColor = DEFAULT_RANGE_COLOR; // translucent; new ranges get this color
  const cellOpts = {
    fill: (f: StreamPolygon) => f.properties.color,
    id: (f: StreamPolygon) => f.properties.id,
    clipTo: "land", // only show ranges over land
    pickable: false, // huge streamed layer — skip the hit index to save memory
  };
  let cells: LayerHandle<StreamPolygon> = map.layer("cells", [], cellOpts);
  map.render();

  const stats = createStatsOverlay(host);
  let count = 0;
  let appendMs = 0;

  let seed = 1;
  let total = DATA_SIZE_TOTALS[String(options.size)] ?? 1_000_000;
  const ctrl = new StreamController<StreamPolygon>({
    source: (o) => makeStreamingPolygons({ total, ...o }), // large ranges (default size)
    onBatch: (batch) => {
      for (const f of batch) {
        f.properties.color = currentColor;
        retained.push(f); // loop, not push(...spread): batch can be up to 1M
      }
      const t0 = performance.now();
      cells.append(batch);
      appendMs += performance.now() - t0;
      count += batch.length;
      stats.update(count, total, count / (appendMs / 1000), ctrl.batchSize);
    },
    onReset: () => {
      retained.length = 0;
      count = 0;
      appendMs = 0;
      cells = map.layer("cells", [], cellOpts);
      stats.update(0, total, 0, ctrl.batchSize, true);
    },
  });
  // "adaptive" auto-tunes the batch to a frame budget; a number fixes it.
  const applyBatch = (opt: string): void => {
    ctrl.adaptive = opt === "adaptive";
    ctrl.batchSize = ctrl.adaptive ? ADAPTIVE_SEED_BATCH : Number(opt);
  };
  // Stream runs only when the user wants it AND the canvas is on-screen (the harness
  // calls setVisible). A manual pause persists across scroll-out/in.
  let visible = true;
  let userRunning = options.stream === "run";
  let lastBatchOpt = String(options.batch);
  applyBatch(lastBatchOpt);
  ctrl.delayMs = Number(options.rate);
  ctrl.setRunning(userRunning && visible);
  ctrl.restart(seed);

  let lastRandomize = Number(options.randomize) || 0;
  let lastRestart = Number(options.restart) || 0;

  return {
    engine: map,
    render: (o) => {
      userRunning = o.stream === "run";
      ctrl.setRunning(userRunning && visible);
      if (o.randomize !== lastRandomize) {
        lastRandomize = Number(o.randomize) || 0;
        currentColor = randomHsl(RANGE_ALPHA); // keep ranges translucent
        for (const f of retained) f.properties.color = currentColor;
        cells.recolor();
      }
      const batchOpt = String(o.batch);
      const rate = Number(o.rate);
      const newTotal = DATA_SIZE_TOTALS[String(o.size)] ?? total;
      if (
        batchOpt !== lastBatchOpt ||
        rate !== ctrl.delayMs ||
        newTotal !== total ||
        o.restart !== lastRestart
      ) {
        lastRestart = Number(o.restart) || 0;
        lastBatchOpt = batchOpt;
        applyBatch(batchOpt);
        ctrl.delayMs = rate;
        total = newTotal;
        ctrl.restart(++seed);
      }
    },
    setVisible: (v) => {
      visible = v;
      ctrl.setRunning(userRunning && visible); // pause offscreen; resume only if user wants run
    },
    dispose: () => {
      ctrl.dispose();
      stats.destroy();
    },
  };
};

geo-data.ts

geo-data.ts

import { geoGraticule } from "d3-geo";
import type { Feature, FeatureCollection, LineString, MultiLineString, MultiPoint, MultiPolygon, Point, Polygon } from "geojson";
import { feature } from "topojson-client";
import land110m from "world-atlas/land-110m.json";

/** A synthetic grid cell with a continuous value and a categorical bioregion. */
export interface Cell {
  id: string;
  geometry: Polygon;
  /** Cell centroid [lon, lat]. */
  center: [number, number];
  /** Continuous field in [0, 1] (value field). */
  value: number;
  /** Categorical bioregion id in 0..7. */
  bioregion: number;
}

/** Base cell size in degrees; the example scales it by powers of two via a slider. */
export const BASE_STEP = 1;

function clamp01(x: number): number {
  return Math.max(0, Math.min(1, x));
}

/** Generate a global grid of `step`°×`step`° cells with smooth synthetic fields. */
export function makeCells(step: number = BASE_STEP): Cell[] {
  const cells: Cell[] = [];
  let col = 0;
  for (let lon = -180; lon < 180; lon += step, col++) {
    let row = 0;
    for (let lat = -90; lat < 90; lat += step, row++) {
      const lonR = (lon * Math.PI) / 180;
      const latR = (lat * Math.PI) / 180;
      const value = clamp01(0.5 + 0.5 * Math.sin(lonR * 2) * Math.cos(latR * 3));
      const field = (Math.sin(lon / 40) + Math.cos(lat / 30)) * 0.5 + 1; // ~[0,2]
      const bioregion = Math.min(7, Math.max(0, Math.floor((field / 2) * 8)));
      const geometry: Polygon = {
        type: "Polygon",
        coordinates: [
          [
            [lon, lat],
            [lon, lat + step],
            [lon + step, lat + step],
            [lon + step, lat],
            [lon, lat],
          ],
        ],
      };
      cells.push({
        id: `${col}-${row}`,
        geometry,
        center: [lon + step / 2, lat + step / 2],
        value,
        bioregion,
      });
    }
  }
  return cells;
}

/** A fine grid over the central third of the globe (lon ±60°, lat ±30°), 4° cells —
 *  a "dense" demo layer (used clipped to land). */
export function centreCells(): Cell[] {
  return makeCells(4).filter((c) => Math.abs(c.center[0]) <= 60 && Math.abs(c.center[1]) <= 30);
}

/** Wrap cells as a FeatureCollection for projection fitting. */
export function cellsToFeatureCollection(cells: readonly Cell[]): FeatureCollection {
  const features: Feature[] = cells.map((c) => ({
    type: "Feature",
    properties: { id: c.id },
    geometry: c.geometry,
  }));
  return { type: "FeatureCollection", features };
}

/** A GeoJSON object d3-geo can fill that isn't part of the strict GeoJSON spec. */
export type Sphere = { type: "Sphere" };

/** The land outline (Natural Earth 110m) plus a sphere to fill as ocean. */
export interface World {
  sphere: Sphere;
  land: MultiPolygon;
}

// Derive the topojson Topology type from feature()'s own signature so we don't
// take a direct dependency on the (transitive) topojson-specification types.
type Topology = Parameters<typeof feature>[0];

/**
 * Convert the bundled world-atlas TopoJSON into a land MultiPolygon and a sphere.
 */
export function loadWorld(): World {
  const topo = land110m as unknown as Topology;
  const fc = feature(topo, topo.objects.land!) as unknown as FeatureCollection<MultiPolygon>;
  return { sphere: { type: "Sphere" }, land: fc.features[0]!.geometry };
}

/** A few well-known cities to show point geometry rendered alongside the grid. */
export interface City {
  id: string;
  name: string;
  geometry: Point;
}

export function makeCities(): City[] {
  const places: [string, number, number][] = [
    ["London", -0.13, 51.51],
    ["New York", -74.01, 40.71],
    ["Tokyo", 139.69, 35.69],
    ["Sydney", 151.21, -33.87],
    ["Cape Town", 18.42, -33.92],
    ["Rio de Janeiro", -43.2, -22.91],
    ["Nairobi", 36.82, -1.29],
    ["Mumbai", 72.88, 19.08],
  ];
  return places.map(([name, lon, lat]) => ({
    id: name!,
    name: name!,
    geometry: { type: "Point", coordinates: [lon!, lat!] },
  }));
}

/** A 20° graticule as one MultiLineString feature. */
export function makeGraticule(): Feature<MultiLineString> {
  return { type: "Feature", properties: {}, geometry: geoGraticule().step([20, 20])() };
}

/** A great-circle-ish route as a LineString feature (London -> New York -> Tokyo). */
export function makeRoute(): Feature<LineString> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "LineString", coordinates: [[-0.13, 51.51], [-74.01, 40.71], [139.69, 35.69]] },
  };
}

/** A cluster of locations as one MultiPoint feature. */
export function makeCluster(): Feature<MultiPoint> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "MultiPoint", coordinates: [[18.42, -33.92], [151.21, -33.87], [-43.2, -22.91], [36.82, -1.29], [72.88, 19.08]] },
  };
}

/** A standalone Polygon feature (a box over the Sahara) to showcase polygon geometry.
 *  Wound CLOCKWISE so d3-geo fills the small box (not the sphere complement). */
export function makeDemoPolygon(): Feature<Polygon> {
  return {
    type: "Feature", properties: { name: "demo-region" },
    geometry: { type: "Polygon", coordinates: [[[0, 15], [0, 30], [30, 30], [30, 15], [0, 15]]] },
  };
}

/** A handful of major rivers as rough named polylines (the bundled world-atlas data
 *  has no rivers), shown on the GeoJSON-features map and used as streaming cluster
 *  centers. Coordinates are approximate [lon, lat] traces, mouth → source. */
export function makeMajorRivers(): Feature<LineString, { name: string }>[] {
  const rivers: [string, [number, number][]][] = [
    ["Amazon", [[-50.0, -0.7], [-55.5, -2.5], [-60.0, -3.1], [-67.9, -3.5], [-73.2, -4.5]]],
    ["Nile", [[31.3, 31.4], [32.9, 24.1], [32.5, 15.6], [32.5, 9.5], [31.6, 2.3]]],
    ["Mississippi", [[-89.2, 29.2], [-90.1, 32.3], [-90.2, 38.6], [-91.2, 43.5], [-95.0, 47.2]]],
    ["Yangtze", [[121.8, 31.4], [114.3, 30.6], [106.5, 29.6], [100.2, 26.9], [94.7, 33.5]]],
    ["Congo", [[12.4, -6.0], [16.2, -4.3], [20.0, -1.0], [25.2, 0.5], [27.2, 3.0]]],
    ["Volga", [[48.0, 46.3], [45.0, 48.7], [44.5, 51.6], [47.5, 54.3], [37.0, 57.3]]],
    ["Ganges", [[90.5, 22.5], [88.0, 24.5], [83.0, 25.4], [78.0, 26.5], [78.9, 30.1]]],
  ];
  return rivers.map(([name, coordinates]) => ({
    type: "Feature",
    properties: { name },
    geometry: { type: "LineString", coordinates },
  }));
}

// ---------------------------------------------------------------------------
// Streaming sources — async generators that emit batches of features lazily
// (only `batchSize` are materialized per tick, never the whole `total`), so a
// consumer can `await`-iterate and append them live. Points/cells are CLUSTERED
// around cities + major-river vertices (not uniform), which the world-map
// examples then clip to land. Used by the "streaming data" examples.
// ---------------------------------------------------------------------------

/** A solid default color all streamed features start with (the example's
 *  "randomize" button swaps in a new color for new + retained features). */
export const DEFAULT_STREAM_COLOR = "#e23b2f";

/** Per-feature properties: a stable id (continues across batches) + a color the
 *  example owns (constant by default; swapped by the randomize button). */
export interface StreamProps {
  id: number;
  color: string;
}
export type StreamPoint = Feature<Point, StreamProps>;
export type StreamPolygon = Feature<Polygon, StreamProps>;

export interface StreamOptions {
  /** Total features emitted before the generator completes. Default 10,000,000. */
  total?: number;
  /** Features per yielded batch. A function is re-read every batch, so the caller can
   *  resize adaptively (e.g. to fit a frame budget). Default 1000. */
  batchSize?: number | (() => number);
  /** Artificial delay between batches (ms), to mirror loading from a file/network.
   *  Even 0 yields a macrotask so the browser can paint between batches. Default 0. */
  delayMs?: number;
  /** Seed for the deterministic PRNG (reproducible streams). Default 1. */
  seed?: number;
  /** Cooperative cancellation: iteration stops once `signal.aborted` is true. */
  signal?: { aborted: boolean };
}

/** Small, fast, seedable PRNG (mulberry32) — deterministic so streams reproduce. */
function mulberry32(seed: number): () => number {
  let a = seed >>> 0;
  return () => {
    a |= 0;
    a = (a + 0x6d2b79f5) | 0;
    let t = Math.imul(a ^ (a >>> 15), 1 | a);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
  };
}

const clamp = (x: number, lo: number, hi: number): number => Math.max(lo, Math.min(hi, x));

/**
 * A clustered point process (Thomas-like): many weighted "hotspot" parents at a
 * range of scales — tight, dense cities; medium clumps along rivers; and a field
 * of random global hotspots with varied spread/weight. Offspring fall around a
 * parent with an exponential (heavy-tailed) radius, so the result is patchy and
 * multi-scale — far closer to real species-occurrence density than a smooth
 * gaussian. Clip-to-land then carves the continental outline.
 */
interface Parent {
  lon: number;
  lat: number;
  /** Mean offspring radius in degrees. */
  spread: number;
  /** Relative share of points drawn from this parent. */
  weight: number;
}

function buildParents(rng: () => number): Parent[] {
  const parents: Parent[] = [];
  for (const c of makeCities())
    parents.push({ lon: c.geometry.coordinates[0]!, lat: c.geometry.coordinates[1]!, spread: 1.5, weight: 3 });
  for (const river of makeMajorRivers())
    for (const p of river.geometry.coordinates) parents.push({ lon: p[0]!, lat: p[1]!, spread: 2.5, weight: 2 });
  // Random global hotspots: rng()*rng() biases toward small spreads/weights, so
  // most clumps are tight with a few diffuse ones — a wide range of scales.
  for (let i = 0; i < 240; i++) {
    parents.push({
      lon: rng() * 360 - 180,
      lat: rng() * 160 - 80,
      spread: 0.6 + 9 * rng() * rng(),
      weight: 0.2 + 3 * rng() * rng(),
    });
  }
  return parents;
}

function cumulativeWeights(parents: readonly Parent[]): number[] {
  const cum: number[] = [];
  let s = 0;
  for (const p of parents) {
    s += p.weight;
    cum.push(s);
  }
  return cum;
}

/** Pick a parent by weight (binary search over cumulative weights). */
function pickParent(rng: () => number, parents: readonly Parent[], cum: readonly number[]): Parent {
  const r = rng() * cum[cum.length - 1]!;
  let lo = 0;
  let hi = cum.length - 1;
  while (lo < hi) {
    const mid = (lo + hi) >> 1;
    if (cum[mid]! < r) lo = mid + 1;
    else hi = mid;
  }
  return parents[lo]!;
}

/** Offspring location around a parent: exponential radius, uniform direction. */
function clusteredLonLat(rng: () => number, p: Parent): [number, number] {
  const radius = -p.spread * Math.log(Math.max(1e-9, rng())); // exponential, mean = spread
  const ang = rng() * 2 * Math.PI;
  return [
    clamp(p.lon + radius * Math.cos(ang), -180, 180),
    clamp(p.lat + radius * Math.sin(ang), -90, 90),
  ];
}

/**
 * An irregular, star-convex polygon ring (3–10 vertices, varied per-vertex radius)
 * centered at [clon, clat] with overall extent ≤ ~`size`° — a rough species range.
 * Angles are evenly spaced with bounded jitter so they stay monotonic ⇒ the ring is
 * simple (non-self-intersecting) and closed. Longitude offsets are widened by
 * 1/cos(lat) so ranges don't look squished toward the poles.
 *
 * WINDING: built CLOCKWISE in [lon, lat] (note the NEGATIVE angle). d3-geo fills on
 * the sphere, and a small exterior ring wound counter-clockwise is treated as its
 * complement → it fills the whole map. See `AGENTS.md` and `geo/project.ts`.
 */
function randomRangeRing(rng: () => number, clon: number, clat: number, size: number): [number, number][] {
  const verts = 3 + Math.floor(rng() * 8); // 3..10
  // Strongly heavy-tailed size: the vast majority of ranges are TINY and only ~3% are
  // visibly large. At high counts the translucent fill then reads as a density gradient
  // (clustered richness hotspots) instead of saturating the whole map red.
  const base =
    rng() < 0.03
      ? size * (0.1 + 0.15 * rng()) // ~3% larger ranges: 0.10..0.25 * size
      : size * (0.02 + 0.07 * rng() * rng()); // most tiny: 0.02..0.09 * size, biased small (rng²)
  const latScale = 1 / Math.max(0.25, Math.cos((clat * Math.PI) / 180));
  const ring: [number, number][] = [];
  for (let i = 0; i < verts; i++) {
    const ang = -((i + 0.5 * rng()) / verts) * 2 * Math.PI; // NEGATIVE ⇒ clockwise ⇒ fills interior
    const r = base * (0.5 + rng()); // per-vertex radius variation
    ring.push([
      clamp(clon + r * Math.cos(ang) * latScale, -180, 180),
      clamp(clat + r * Math.sin(ang), -90, 90),
    ]);
  }
  ring.push(ring[0]!); // close the ring
  return ring;
}

const tick = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));

/** Stream world points as GeoJSON `Feature<Point>` batches, clustered around
 *  many multi-scale hotspots. All start with DEFAULT_STREAM_COLOR. */
export async function* makeStreamingPoints(opts: StreamOptions = {}): AsyncGenerator<StreamPoint[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPoint[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [lon, lat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Point", coordinates: [lon, lat] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

/** Stream irregular polygon "ranges" clustered around hotspots — each a 3–10
 *  vertex star-convex polygon of varied size ≤ ~`size`°, to mimic species ranges.
 *  All start with DEFAULT_STREAM_COLOR; the example renders them very transparent
 *  so overlapping ranges build up richness. */
export async function* makeStreamingPolygons(
  opts: StreamOptions & { size?: number } = {},
): AsyncGenerator<StreamPolygon[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, size = 16, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPolygon[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [clon, clat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Polygon", coordinates: [randomRangeRing(rng, clon, clat, size)] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

streaming.ts

streaming.ts

/**
 * Drives a streaming source (an async batch generator) into a consumer, with
 * run/pause and restart — the plumbing shared by the streaming examples so each
 * `draw.ts` only has to say HOW to append a batch (the d3gl-specific part).
 */

export interface StreamSourceOpts {
  /** Re-read every batch (the controller may resize it adaptively). */
  batchSize: () => number;
  delayMs: number;
  seed: number;
  signal: { aborted: boolean };
}

export interface StreamControllerOpts<T> {
  /** Build a fresh source generator for a (re)started session. */
  source: (o: StreamSourceOpts) => AsyncGenerator<T[]>;
  /** Consume one freshly-arrived batch (append it to the layer + retain it). */
  onBatch: (batch: T[]) => void;
  /** Clear retained state + the layer at the start of a (re)started session. */
  onReset: () => void;
}

/**
 * One streaming session at a time. `restart(seed)` resets and starts a new
 * session (superseding any running one); `setRunning(false)` pauses the pump;
 * `dispose()` stops everything for good.
 *
 * Adaptive mode (`adaptive = true`): after each batch the controller times how long
 * `onBatch` (build + append + paint) took and nudges `batchSize` toward the number
 * of features that fit `frameBudgetMs` — so the stream runs as fast as the backend
 * allows while staying responsive (the old Bioregions "fit ~30fps" trick). The source
 * re-reads `batchSize` every batch via the getter passed in `restart`.
 */
export class StreamController<T> {
  batchSize = 1000;
  delayMs = 0;
  /** When true, `batchSize` is auto-tuned each batch to fit `frameBudgetMs`. */
  adaptive = false;
  /** Per-batch time budget for adaptive mode (≈30fps). */
  frameBudgetMs = 32;
  private session = 0;
  private running = true;
  private disposed = false;

  constructor(private readonly opts: StreamControllerOpts<T>) {}

  setRunning(v: boolean): void {
    this.running = v;
  }

  /** Reset + start a fresh session (supersedes any in flight). */
  restart(seed: number): void {
    if (this.disposed) return;
    this.opts.onReset();
    void this.pump(seed);
  }

  /** Stop the current session permanently (call from the example's dispose). */
  dispose(): void {
    this.disposed = true;
    this.session++;
  }

  private async pump(seed: number): Promise<void> {
    const my = ++this.session;
    const signal = { aborted: false };
    // Getter so adaptive resizes take effect on the very next batch the source pulls.
    const gen = this.opts.source({ batchSize: () => this.batchSize, delayMs: this.delayMs, seed, signal });
    for await (const batch of gen) {
      // Superseded (restart) or disposed → abort this session.
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      // Paused → idle without consuming, until resumed / superseded / disposed.
      while (!this.running && !this.disposed && my === this.session) {
        await new Promise((r) => setTimeout(r, 60));
      }
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      const t0 = performance.now();
      this.opts.onBatch(batch);
      if (this.adaptive) {
        // Steer toward the budget: scale by budget/elapsed, clamped to ≤2× per step so
        // it ramps quickly but doesn't oscillate. Bounded to [1, 1_000_000].
        const dt = Math.max(0.5, performance.now() - t0);
        const factor = Math.max(0.5, Math.min(2, this.frameBudgetMs / dt));
        this.batchSize = Math.max(1, Math.min(1_000_000, Math.round(this.batchSize * factor)));
      }
    }
  }
}

/** A random vivid color (used by the "randomize colors" button). Pass `alpha` < 1
 *  for translucent fills (e.g. overlapping polygon ranges). */
export function randomHsl(alpha = 1): string {
  const h = Math.floor(Math.random() * 360);
  return alpha >= 1 ? `hsl(${h}, 70%, 55%)` : `hsla(${h}, 70%, 55%, ${alpha})`;
}

/** The batch-size choices offered by the streaming examples' control. "adaptive"
 *  auto-tunes the batch to fit a frame budget (see StreamController.adaptive). */
export const BATCH_SIZES = ["adaptive", "1", "10", "100", "1000", "100000", "1000000"];
/** Seed batch size when adaptive mode starts, before it converges. */
export const ADAPTIVE_SEED_BATCH = 256;
/** The artificial per-batch delay choices (ms) — mirrors loading from a file. */
export const RATES_MS = ["0", "16", "100", "500"];
/** Data-size choices (total features to stream) and their numeric totals. */
export const DATA_SIZES = ["100k", "1M", "10M"];
export const DATA_SIZE_TOTALS: Record<string, number> = {
  "100k": 100_000,
  "1M": 1_000_000,
  "10M": 10_000_000,
};

stats-overlay.ts

stats-overlay.ts

/**
 * A tiny HTML overlay (top-right of the canvas) showing how many records have
 * streamed in — count, percentage of the total, and the average append speed so
 * far (records / time spent in `append`). Mirrors the Infomap Bioregions readout.
 */
export interface StatsOverlay {
  /** Update the readout. Throttled to ~10 Hz unless `force`. `batch` (optional) shows
   *  the current batch size — handy to watch adaptive batching converge. */
  update(count: number, total: number, recordsPerSec: number, batch?: number, force?: boolean): void;
  destroy(): void;
}

export function createStatsOverlay(host: HTMLElement): StatsOverlay {
  const el = document.createElement("div");
  el.className =
    "absolute top-2 right-2 pointer-events-none rounded-md bg-white/80 px-2 py-1 " +
    "font-mono text-[11px] leading-tight text-[#333] shadow-sm [font-variant-numeric:tabular-nums]";
  host.appendChild(el);

  const fmt = (n: number): string => Math.round(n).toLocaleString();
  let last = 0;

  return {
    update(count, total, recordsPerSec, batch, force = false) {
      const now = performance.now();
      if (!force && now - last < 100) return;
      last = now;
      const ratio = total > 0 ? count / total : 0;
      const pct = (ratio * 100).toFixed(ratio < 0.1 ? 1 : 0);
      const batchLine = batch != null ? `<br>batch <b>${fmt(batch)}</b>` : "";
      el.innerHTML = `records <b>${fmt(count)}</b> (${pct}%)<br>speed <b>${fmt(recordsPerSec)}</b> rec/s${batchLine}`;
    },
    destroy() {
      el.remove();
    },
  };
}

Streaming points on a scatter plot

The same point source, plotted with x = longitude, y = −latitude so the familiar world shape emerges — Plot.points().append() is the same incremental append as the map’s GeoMap.layer().append().

autowebglcanvassvg

Export PNG

fps 0frame 0 ms

Stream

runpause

RestartRandomize colors

draw.ts

draw.ts

import { plot, type LayerHandle } from "@mapequation/d3gl/map";
import { makeStreamingPoints, DEFAULT_STREAM_COLOR, type StreamPoint } from "../shared/geo-data.js";
import { StreamController, randomHsl, DATA_SIZE_TOTALS, ADAPTIVE_SEED_BATCH } from "../shared/streaming.js";
import { createStatsOverlay } from "../shared/stats-overlay.js";
import type { ImperativeSetup } from "../types.js";

/**
 * The same streaming point source as the world-map example, but plotted on a
 * scatter chart: x = longitude, y = −latitude (so north is up). Reusing
 * `makeStreamingPoints` shows `Plot.points().append()` is the same incremental
 * append as the map's `GeoMap.layer().append()`. Points start red; "Randomize
 * colors" recolors future + retained points.
 */
export const setup: ImperativeSetup = (host, { width, height, backend, options }) => {
  const chart = plot(host, { width, height, backend });
  // Fit lon∈[-180,180], lat∈[-90,90] into the canvas (x=lon, y=−lat, 0,0 centered).
  const k = Math.min(width / 360, height / 180) * 0.92;
  chart.setTransform({ k, x: width / 2, y: height / 2 });
  chart.enableZoom([0.5, 40]);

  const retained: StreamPoint[] = [];
  let currentColor = DEFAULT_STREAM_COLOR;
  const pointOpts = {
    x: (f: StreamPoint) => f.geometry.coordinates[0], // longitude
    y: (f: StreamPoint) => -f.geometry.coordinates[1], // −latitude (north up)
    radius: 1,
    fill: (f: StreamPoint) => f.properties.color,
    id: (f: StreamPoint) => f.properties.id,
    sizeMode: "screen" as const, // constant 1px dots regardless of zoom
    pickable: false, // huge streamed layer — skip the hit index to save memory
  };
  let points: LayerHandle<StreamPoint> = chart.points("points", [], pointOpts);
  chart.render();

  const stats = createStatsOverlay(host);
  let count = 0;
  let appendMs = 0;

  let seed = 1;
  let total = DATA_SIZE_TOTALS[String(options.size)] ?? 1_000_000;
  const ctrl = new StreamController<StreamPoint>({
    source: (o) => makeStreamingPoints({ total, ...o }),
    onBatch: (batch) => {
      for (const f of batch) {
        f.properties.color = currentColor;
        retained.push(f); // loop, not push(...spread): batch can be up to 1M
      }
      const t0 = performance.now();
      points.append(batch);
      appendMs += performance.now() - t0;
      count += batch.length;
      stats.update(count, total, count / (appendMs / 1000), ctrl.batchSize);
    },
    onReset: () => {
      retained.length = 0;
      count = 0;
      appendMs = 0;
      points = chart.points("points", [], pointOpts);
      stats.update(0, total, 0, ctrl.batchSize, true);
    },
  });
  // "adaptive" auto-tunes the batch to a frame budget; a number fixes it.
  const applyBatch = (opt: string): void => {
    ctrl.adaptive = opt === "adaptive";
    ctrl.batchSize = ctrl.adaptive ? ADAPTIVE_SEED_BATCH : Number(opt);
  };
  // Stream runs only when the user wants it AND the canvas is on-screen (the harness
  // calls setVisible). A manual pause persists across scroll-out/in.
  let visible = true;
  let userRunning = options.stream === "run";
  let lastBatchOpt = String(options.batch);
  applyBatch(lastBatchOpt);
  ctrl.delayMs = Number(options.rate);
  ctrl.setRunning(userRunning && visible);
  ctrl.restart(seed);

  let lastRandomize = Number(options.randomize) || 0;
  let lastRestart = Number(options.restart) || 0;

  return {
    engine: chart,
    render: (o) => {
      userRunning = o.stream === "run";
      ctrl.setRunning(userRunning && visible);
      if (o.randomize !== lastRandomize) {
        lastRandomize = Number(o.randomize) || 0;
        currentColor = randomHsl();
        for (const f of retained) f.properties.color = currentColor;
        points.recolor();
      }
      const batchOpt = String(o.batch);
      const rate = Number(o.rate);
      const newTotal = DATA_SIZE_TOTALS[String(o.size)] ?? total;
      if (
        batchOpt !== lastBatchOpt ||
        rate !== ctrl.delayMs ||
        newTotal !== total ||
        o.restart !== lastRestart
      ) {
        lastRestart = Number(o.restart) || 0;
        lastBatchOpt = batchOpt;
        applyBatch(batchOpt);
        ctrl.delayMs = rate;
        total = newTotal;
        ctrl.restart(++seed);
      }
    },
    setVisible: (v) => {
      visible = v;
      ctrl.setRunning(userRunning && visible); // pause offscreen; resume only if user wants run
    },
    dispose: () => {
      ctrl.dispose();
      stats.destroy();
    },
  };
};

geo-data.ts

geo-data.ts

import { geoGraticule } from "d3-geo";
import type { Feature, FeatureCollection, LineString, MultiLineString, MultiPoint, MultiPolygon, Point, Polygon } from "geojson";
import { feature } from "topojson-client";
import land110m from "world-atlas/land-110m.json";

/** A synthetic grid cell with a continuous value and a categorical bioregion. */
export interface Cell {
  id: string;
  geometry: Polygon;
  /** Cell centroid [lon, lat]. */
  center: [number, number];
  /** Continuous field in [0, 1] (value field). */
  value: number;
  /** Categorical bioregion id in 0..7. */
  bioregion: number;
}

/** Base cell size in degrees; the example scales it by powers of two via a slider. */
export const BASE_STEP = 1;

function clamp01(x: number): number {
  return Math.max(0, Math.min(1, x));
}

/** Generate a global grid of `step`°×`step`° cells with smooth synthetic fields. */
export function makeCells(step: number = BASE_STEP): Cell[] {
  const cells: Cell[] = [];
  let col = 0;
  for (let lon = -180; lon < 180; lon += step, col++) {
    let row = 0;
    for (let lat = -90; lat < 90; lat += step, row++) {
      const lonR = (lon * Math.PI) / 180;
      const latR = (lat * Math.PI) / 180;
      const value = clamp01(0.5 + 0.5 * Math.sin(lonR * 2) * Math.cos(latR * 3));
      const field = (Math.sin(lon / 40) + Math.cos(lat / 30)) * 0.5 + 1; // ~[0,2]
      const bioregion = Math.min(7, Math.max(0, Math.floor((field / 2) * 8)));
      const geometry: Polygon = {
        type: "Polygon",
        coordinates: [
          [
            [lon, lat],
            [lon, lat + step],
            [lon + step, lat + step],
            [lon + step, lat],
            [lon, lat],
          ],
        ],
      };
      cells.push({
        id: `${col}-${row}`,
        geometry,
        center: [lon + step / 2, lat + step / 2],
        value,
        bioregion,
      });
    }
  }
  return cells;
}

/** A fine grid over the central third of the globe (lon ±60°, lat ±30°), 4° cells —
 *  a "dense" demo layer (used clipped to land). */
export function centreCells(): Cell[] {
  return makeCells(4).filter((c) => Math.abs(c.center[0]) <= 60 && Math.abs(c.center[1]) <= 30);
}

/** Wrap cells as a FeatureCollection for projection fitting. */
export function cellsToFeatureCollection(cells: readonly Cell[]): FeatureCollection {
  const features: Feature[] = cells.map((c) => ({
    type: "Feature",
    properties: { id: c.id },
    geometry: c.geometry,
  }));
  return { type: "FeatureCollection", features };
}

/** A GeoJSON object d3-geo can fill that isn't part of the strict GeoJSON spec. */
export type Sphere = { type: "Sphere" };

/** The land outline (Natural Earth 110m) plus a sphere to fill as ocean. */
export interface World {
  sphere: Sphere;
  land: MultiPolygon;
}

// Derive the topojson Topology type from feature()'s own signature so we don't
// take a direct dependency on the (transitive) topojson-specification types.
type Topology = Parameters<typeof feature>[0];

/**
 * Convert the bundled world-atlas TopoJSON into a land MultiPolygon and a sphere.
 */
export function loadWorld(): World {
  const topo = land110m as unknown as Topology;
  const fc = feature(topo, topo.objects.land!) as unknown as FeatureCollection<MultiPolygon>;
  return { sphere: { type: "Sphere" }, land: fc.features[0]!.geometry };
}

/** A few well-known cities to show point geometry rendered alongside the grid. */
export interface City {
  id: string;
  name: string;
  geometry: Point;
}

export function makeCities(): City[] {
  const places: [string, number, number][] = [
    ["London", -0.13, 51.51],
    ["New York", -74.01, 40.71],
    ["Tokyo", 139.69, 35.69],
    ["Sydney", 151.21, -33.87],
    ["Cape Town", 18.42, -33.92],
    ["Rio de Janeiro", -43.2, -22.91],
    ["Nairobi", 36.82, -1.29],
    ["Mumbai", 72.88, 19.08],
  ];
  return places.map(([name, lon, lat]) => ({
    id: name!,
    name: name!,
    geometry: { type: "Point", coordinates: [lon!, lat!] },
  }));
}

/** A 20° graticule as one MultiLineString feature. */
export function makeGraticule(): Feature<MultiLineString> {
  return { type: "Feature", properties: {}, geometry: geoGraticule().step([20, 20])() };
}

/** A great-circle-ish route as a LineString feature (London -> New York -> Tokyo). */
export function makeRoute(): Feature<LineString> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "LineString", coordinates: [[-0.13, 51.51], [-74.01, 40.71], [139.69, 35.69]] },
  };
}

/** A cluster of locations as one MultiPoint feature. */
export function makeCluster(): Feature<MultiPoint> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "MultiPoint", coordinates: [[18.42, -33.92], [151.21, -33.87], [-43.2, -22.91], [36.82, -1.29], [72.88, 19.08]] },
  };
}

/** A standalone Polygon feature (a box over the Sahara) to showcase polygon geometry.
 *  Wound CLOCKWISE so d3-geo fills the small box (not the sphere complement). */
export function makeDemoPolygon(): Feature<Polygon> {
  return {
    type: "Feature", properties: { name: "demo-region" },
    geometry: { type: "Polygon", coordinates: [[[0, 15], [0, 30], [30, 30], [30, 15], [0, 15]]] },
  };
}

/** A handful of major rivers as rough named polylines (the bundled world-atlas data
 *  has no rivers), shown on the GeoJSON-features map and used as streaming cluster
 *  centers. Coordinates are approximate [lon, lat] traces, mouth → source. */
export function makeMajorRivers(): Feature<LineString, { name: string }>[] {
  const rivers: [string, [number, number][]][] = [
    ["Amazon", [[-50.0, -0.7], [-55.5, -2.5], [-60.0, -3.1], [-67.9, -3.5], [-73.2, -4.5]]],
    ["Nile", [[31.3, 31.4], [32.9, 24.1], [32.5, 15.6], [32.5, 9.5], [31.6, 2.3]]],
    ["Mississippi", [[-89.2, 29.2], [-90.1, 32.3], [-90.2, 38.6], [-91.2, 43.5], [-95.0, 47.2]]],
    ["Yangtze", [[121.8, 31.4], [114.3, 30.6], [106.5, 29.6], [100.2, 26.9], [94.7, 33.5]]],
    ["Congo", [[12.4, -6.0], [16.2, -4.3], [20.0, -1.0], [25.2, 0.5], [27.2, 3.0]]],
    ["Volga", [[48.0, 46.3], [45.0, 48.7], [44.5, 51.6], [47.5, 54.3], [37.0, 57.3]]],
    ["Ganges", [[90.5, 22.5], [88.0, 24.5], [83.0, 25.4], [78.0, 26.5], [78.9, 30.1]]],
  ];
  return rivers.map(([name, coordinates]) => ({
    type: "Feature",
    properties: { name },
    geometry: { type: "LineString", coordinates },
  }));
}

// ---------------------------------------------------------------------------
// Streaming sources — async generators that emit batches of features lazily
// (only `batchSize` are materialized per tick, never the whole `total`), so a
// consumer can `await`-iterate and append them live. Points/cells are CLUSTERED
// around cities + major-river vertices (not uniform), which the world-map
// examples then clip to land. Used by the "streaming data" examples.
// ---------------------------------------------------------------------------

/** A solid default color all streamed features start with (the example's
 *  "randomize" button swaps in a new color for new + retained features). */
export const DEFAULT_STREAM_COLOR = "#e23b2f";

/** Per-feature properties: a stable id (continues across batches) + a color the
 *  example owns (constant by default; swapped by the randomize button). */
export interface StreamProps {
  id: number;
  color: string;
}
export type StreamPoint = Feature<Point, StreamProps>;
export type StreamPolygon = Feature<Polygon, StreamProps>;

export interface StreamOptions {
  /** Total features emitted before the generator completes. Default 10,000,000. */
  total?: number;
  /** Features per yielded batch. A function is re-read every batch, so the caller can
   *  resize adaptively (e.g. to fit a frame budget). Default 1000. */
  batchSize?: number | (() => number);
  /** Artificial delay between batches (ms), to mirror loading from a file/network.
   *  Even 0 yields a macrotask so the browser can paint between batches. Default 0. */
  delayMs?: number;
  /** Seed for the deterministic PRNG (reproducible streams). Default 1. */
  seed?: number;
  /** Cooperative cancellation: iteration stops once `signal.aborted` is true. */
  signal?: { aborted: boolean };
}

/** Small, fast, seedable PRNG (mulberry32) — deterministic so streams reproduce. */
function mulberry32(seed: number): () => number {
  let a = seed >>> 0;
  return () => {
    a |= 0;
    a = (a + 0x6d2b79f5) | 0;
    let t = Math.imul(a ^ (a >>> 15), 1 | a);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
  };
}

const clamp = (x: number, lo: number, hi: number): number => Math.max(lo, Math.min(hi, x));

/**
 * A clustered point process (Thomas-like): many weighted "hotspot" parents at a
 * range of scales — tight, dense cities; medium clumps along rivers; and a field
 * of random global hotspots with varied spread/weight. Offspring fall around a
 * parent with an exponential (heavy-tailed) radius, so the result is patchy and
 * multi-scale — far closer to real species-occurrence density than a smooth
 * gaussian. Clip-to-land then carves the continental outline.
 */
interface Parent {
  lon: number;
  lat: number;
  /** Mean offspring radius in degrees. */
  spread: number;
  /** Relative share of points drawn from this parent. */
  weight: number;
}

function buildParents(rng: () => number): Parent[] {
  const parents: Parent[] = [];
  for (const c of makeCities())
    parents.push({ lon: c.geometry.coordinates[0]!, lat: c.geometry.coordinates[1]!, spread: 1.5, weight: 3 });
  for (const river of makeMajorRivers())
    for (const p of river.geometry.coordinates) parents.push({ lon: p[0]!, lat: p[1]!, spread: 2.5, weight: 2 });
  // Random global hotspots: rng()*rng() biases toward small spreads/weights, so
  // most clumps are tight with a few diffuse ones — a wide range of scales.
  for (let i = 0; i < 240; i++) {
    parents.push({
      lon: rng() * 360 - 180,
      lat: rng() * 160 - 80,
      spread: 0.6 + 9 * rng() * rng(),
      weight: 0.2 + 3 * rng() * rng(),
    });
  }
  return parents;
}

function cumulativeWeights(parents: readonly Parent[]): number[] {
  const cum: number[] = [];
  let s = 0;
  for (const p of parents) {
    s += p.weight;
    cum.push(s);
  }
  return cum;
}

/** Pick a parent by weight (binary search over cumulative weights). */
function pickParent(rng: () => number, parents: readonly Parent[], cum: readonly number[]): Parent {
  const r = rng() * cum[cum.length - 1]!;
  let lo = 0;
  let hi = cum.length - 1;
  while (lo < hi) {
    const mid = (lo + hi) >> 1;
    if (cum[mid]! < r) lo = mid + 1;
    else hi = mid;
  }
  return parents[lo]!;
}

/** Offspring location around a parent: exponential radius, uniform direction. */
function clusteredLonLat(rng: () => number, p: Parent): [number, number] {
  const radius = -p.spread * Math.log(Math.max(1e-9, rng())); // exponential, mean = spread
  const ang = rng() * 2 * Math.PI;
  return [
    clamp(p.lon + radius * Math.cos(ang), -180, 180),
    clamp(p.lat + radius * Math.sin(ang), -90, 90),
  ];
}

/**
 * An irregular, star-convex polygon ring (3–10 vertices, varied per-vertex radius)
 * centered at [clon, clat] with overall extent ≤ ~`size`° — a rough species range.
 * Angles are evenly spaced with bounded jitter so they stay monotonic ⇒ the ring is
 * simple (non-self-intersecting) and closed. Longitude offsets are widened by
 * 1/cos(lat) so ranges don't look squished toward the poles.
 *
 * WINDING: built CLOCKWISE in [lon, lat] (note the NEGATIVE angle). d3-geo fills on
 * the sphere, and a small exterior ring wound counter-clockwise is treated as its
 * complement → it fills the whole map. See `AGENTS.md` and `geo/project.ts`.
 */
function randomRangeRing(rng: () => number, clon: number, clat: number, size: number): [number, number][] {
  const verts = 3 + Math.floor(rng() * 8); // 3..10
  // Strongly heavy-tailed size: the vast majority of ranges are TINY and only ~3% are
  // visibly large. At high counts the translucent fill then reads as a density gradient
  // (clustered richness hotspots) instead of saturating the whole map red.
  const base =
    rng() < 0.03
      ? size * (0.1 + 0.15 * rng()) // ~3% larger ranges: 0.10..0.25 * size
      : size * (0.02 + 0.07 * rng() * rng()); // most tiny: 0.02..0.09 * size, biased small (rng²)
  const latScale = 1 / Math.max(0.25, Math.cos((clat * Math.PI) / 180));
  const ring: [number, number][] = [];
  for (let i = 0; i < verts; i++) {
    const ang = -((i + 0.5 * rng()) / verts) * 2 * Math.PI; // NEGATIVE ⇒ clockwise ⇒ fills interior
    const r = base * (0.5 + rng()); // per-vertex radius variation
    ring.push([
      clamp(clon + r * Math.cos(ang) * latScale, -180, 180),
      clamp(clat + r * Math.sin(ang), -90, 90),
    ]);
  }
  ring.push(ring[0]!); // close the ring
  return ring;
}

const tick = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));

/** Stream world points as GeoJSON `Feature<Point>` batches, clustered around
 *  many multi-scale hotspots. All start with DEFAULT_STREAM_COLOR. */
export async function* makeStreamingPoints(opts: StreamOptions = {}): AsyncGenerator<StreamPoint[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPoint[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [lon, lat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Point", coordinates: [lon, lat] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

/** Stream irregular polygon "ranges" clustered around hotspots — each a 3–10
 *  vertex star-convex polygon of varied size ≤ ~`size`°, to mimic species ranges.
 *  All start with DEFAULT_STREAM_COLOR; the example renders them very transparent
 *  so overlapping ranges build up richness. */
export async function* makeStreamingPolygons(
  opts: StreamOptions & { size?: number } = {},
): AsyncGenerator<StreamPolygon[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, size = 16, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPolygon[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [clon, clat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Polygon", coordinates: [randomRangeRing(rng, clon, clat, size)] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

streaming.ts

streaming.ts

/**
 * Drives a streaming source (an async batch generator) into a consumer, with
 * run/pause and restart — the plumbing shared by the streaming examples so each
 * `draw.ts` only has to say HOW to append a batch (the d3gl-specific part).
 */

export interface StreamSourceOpts {
  /** Re-read every batch (the controller may resize it adaptively). */
  batchSize: () => number;
  delayMs: number;
  seed: number;
  signal: { aborted: boolean };
}

export interface StreamControllerOpts<T> {
  /** Build a fresh source generator for a (re)started session. */
  source: (o: StreamSourceOpts) => AsyncGenerator<T[]>;
  /** Consume one freshly-arrived batch (append it to the layer + retain it). */
  onBatch: (batch: T[]) => void;
  /** Clear retained state + the layer at the start of a (re)started session. */
  onReset: () => void;
}

/**
 * One streaming session at a time. `restart(seed)` resets and starts a new
 * session (superseding any running one); `setRunning(false)` pauses the pump;
 * `dispose()` stops everything for good.
 *
 * Adaptive mode (`adaptive = true`): after each batch the controller times how long
 * `onBatch` (build + append + paint) took and nudges `batchSize` toward the number
 * of features that fit `frameBudgetMs` — so the stream runs as fast as the backend
 * allows while staying responsive (the old Bioregions "fit ~30fps" trick). The source
 * re-reads `batchSize` every batch via the getter passed in `restart`.
 */
export class StreamController<T> {
  batchSize = 1000;
  delayMs = 0;
  /** When true, `batchSize` is auto-tuned each batch to fit `frameBudgetMs`. */
  adaptive = false;
  /** Per-batch time budget for adaptive mode (≈30fps). */
  frameBudgetMs = 32;
  private session = 0;
  private running = true;
  private disposed = false;

  constructor(private readonly opts: StreamControllerOpts<T>) {}

  setRunning(v: boolean): void {
    this.running = v;
  }

  /** Reset + start a fresh session (supersedes any in flight). */
  restart(seed: number): void {
    if (this.disposed) return;
    this.opts.onReset();
    void this.pump(seed);
  }

  /** Stop the current session permanently (call from the example's dispose). */
  dispose(): void {
    this.disposed = true;
    this.session++;
  }

  private async pump(seed: number): Promise<void> {
    const my = ++this.session;
    const signal = { aborted: false };
    // Getter so adaptive resizes take effect on the very next batch the source pulls.
    const gen = this.opts.source({ batchSize: () => this.batchSize, delayMs: this.delayMs, seed, signal });
    for await (const batch of gen) {
      // Superseded (restart) or disposed → abort this session.
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      // Paused → idle without consuming, until resumed / superseded / disposed.
      while (!this.running && !this.disposed && my === this.session) {
        await new Promise((r) => setTimeout(r, 60));
      }
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      const t0 = performance.now();
      this.opts.onBatch(batch);
      if (this.adaptive) {
        // Steer toward the budget: scale by budget/elapsed, clamped to ≤2× per step so
        // it ramps quickly but doesn't oscillate. Bounded to [1, 1_000_000].
        const dt = Math.max(0.5, performance.now() - t0);
        const factor = Math.max(0.5, Math.min(2, this.frameBudgetMs / dt));
        this.batchSize = Math.max(1, Math.min(1_000_000, Math.round(this.batchSize * factor)));
      }
    }
  }
}

/** A random vivid color (used by the "randomize colors" button). Pass `alpha` < 1
 *  for translucent fills (e.g. overlapping polygon ranges). */
export function randomHsl(alpha = 1): string {
  const h = Math.floor(Math.random() * 360);
  return alpha >= 1 ? `hsl(${h}, 70%, 55%)` : `hsla(${h}, 70%, 55%, ${alpha})`;
}

/** The batch-size choices offered by the streaming examples' control. "adaptive"
 *  auto-tunes the batch to fit a frame budget (see StreamController.adaptive). */
export const BATCH_SIZES = ["adaptive", "1", "10", "100", "1000", "100000", "1000000"];
/** Seed batch size when adaptive mode starts, before it converges. */
export const ADAPTIVE_SEED_BATCH = 256;
/** The artificial per-batch delay choices (ms) — mirrors loading from a file. */
export const RATES_MS = ["0", "16", "100", "500"];
/** Data-size choices (total features to stream) and their numeric totals. */
export const DATA_SIZES = ["100k", "1M", "10M"];
export const DATA_SIZE_TOTALS: Record<string, number> = {
  "100k": 100_000,
  "1M": 1_000_000,
  "10M": 10_000_000,
};

stats-overlay.ts

stats-overlay.ts

/**
 * A tiny HTML overlay (top-right of the canvas) showing how many records have
 * streamed in — count, percentage of the total, and the average append speed so
 * far (records / time spent in `append`). Mirrors the Infomap Bioregions readout.
 */
export interface StatsOverlay {
  /** Update the readout. Throttled to ~10 Hz unless `force`. `batch` (optional) shows
   *  the current batch size — handy to watch adaptive batching converge. */
  update(count: number, total: number, recordsPerSec: number, batch?: number, force?: boolean): void;
  destroy(): void;
}

export function createStatsOverlay(host: HTMLElement): StatsOverlay {
  const el = document.createElement("div");
  el.className =
    "absolute top-2 right-2 pointer-events-none rounded-md bg-white/80 px-2 py-1 " +
    "font-mono text-[11px] leading-tight text-[#333] shadow-sm [font-variant-numeric:tabular-nums]";
  host.appendChild(el);

  const fmt = (n: number): string => Math.round(n).toLocaleString();
  let last = 0;

  return {
    update(count, total, recordsPerSec, batch, force = false) {
      const now = performance.now();
      if (!force && now - last < 100) return;
      last = now;
      const ratio = total > 0 ? count / total : 0;
      const pct = (ratio * 100).toFixed(ratio < 0.1 ? 1 : 0);
      const batchLine = batch != null ? `<br>batch <b>${fmt(batch)}</b>` : "";
      el.innerHTML = `records <b>${fmt(count)}</b> (${pct}%)<br>speed <b>${fmt(recordsPerSec)}</b> rec/s${batchLine}`;
    },
    destroy() {
      el.remove();
    },
  };
}

Pass-through: uncapped streaming

The examples above retain every feature in d3gl so it can re-project, recolor, and hit-test — which caps out around 4–16M features (storage, not drawing, is the limit). Pass-through mode (passThrough: true) retains nothing in d3gl: you own the data, and d3gl pulls from it, draws, and discards on each repaint. The ceiling becomes whatever your own array costs — 250M+ for a packed Float32Array. Pass features (or points data) as a callback that returns your current array; d3gl re-invokes it on each full repaint, and handle.append(batch) draws new arrivals immediately.

let nodes = [];                                  // you own the data
const layer = map.layer("nodes", () => nodes, {  // callback, re-pulled per repaint
  passThrough: true,
  fill: (f) => f.properties.color,
});
// stream in:
nodes.push(...batch);
layer.append(batch);                             // draws the new batch on top, O(new)

It works for all GeoJSON geometry on both Canvas and WebGL (points → analytic circles; polygons/lines → projected paths, re-tessellated per settle on WebGL). The trade-off:

	Retained (default)	Pass-through
Ceiling	~4–16M	your array (250M+)
Memory in d3gl	~130–480 B/feature	~0
During pan/zoom	always crisp	slightly stale raster, re-crisp on settle
Picking (pick/hover)	yes	no
Per-feature recolor / mutate	yes	re-pull from your data
Clip to another layer (clipTo)	yes	not yet (ignored)
Backends	all (incl. SVG)	Canvas + WebGL (SVG rejects it)

Use retained for datasets up to a few million where you need crisp interaction, hit-testing, or per-feature updates. Use pass-through for huge or live-streaming sets where a brief raster during gestures is fine and picking isn’t needed. The demo below streams to 10M points with a flat memory footprint (default auto backend → upgrades to WebGL).

autowebglcanvassvg

Export PNG

fps 0frame 0 ms

Stream

runpause

RestartRandomize colors

draw.ts

draw.ts

import { geoEquirectangular } from "d3-geo";
import { geoMap, type LayerHandle } from "@mapequation/d3gl/map";
import {
  loadWorld,
  makeStreamingPoints,
  DEFAULT_STREAM_COLOR,
  type StreamPoint,
} from "../shared/geo-data.js";
import { StreamController, randomHsl, DATA_SIZE_TOTALS, ADAPTIVE_SEED_BATCH } from "../shared/streaming.js";
import { createStatsOverlay } from "../shared/stats-overlay.js";
import type { ImperativeSetup } from "../types.js";

const OCEAN = "#dbe7f3";
const LAND = "#e9e7df";

/**
 * Stream points (clustered around cities + rivers) using the pass-through path
 * (`passThrough: true`). The engine re-reads `() => retained` each repaint so the
 * full point set re-projects on pan/zoom — flat GPU memory, no retained ceiling.
 * New batches are also drawn immediately via `points.append(batch)`.
 *
 * Contrast with the `streaming-points` example (retained path): that example caps
 * at ~4–16M and builds a hit index; this one is uncapped, not pickable, and shows
 * a slightly stale raster during pan/zoom (re-crisps on settle).
 */
export const setup: ImperativeSetup = (host, { width, height, backend, options }) => {
  const world = loadWorld();
  const projection = geoEquirectangular().fitSize([width, height], { type: "Sphere" });
  const map = geoMap(host, { width, height, projection, backend });
  map.layer("ocean", [world.sphere], { fill: OCEAN });
  map.layer("land", [world.land], { fill: LAND, stroke: "#9aa3ad", lineWidth: 0.4 });
  map.enableZoom([1, 40]);

  const retained: StreamPoint[] = [];
  let currentColor = DEFAULT_STREAM_COLOR; // the color new points get (until randomized)
  const pointOpts = {
    fill: (f: StreamPoint) => f.properties.color,
    pointRadius: 0.5,
    // pass-through layers are not pickable (no hit index); no id/clipTo needed
  };
  // passThrough: true — engine re-invokes `() => retained` each repaint (pan/zoom)
  let points: LayerHandle<StreamPoint> = map.layer("points", () => retained, { passThrough: true, ...pointOpts });
  map.render();

  const stats = createStatsOverlay(host);
  let count = 0; // records appended this session
  let appendMs = 0; // total time spent in append() — gives the average speed

  let seed = 1;
  let total = DATA_SIZE_TOTALS[String(options.size)] ?? 10_000_000;
  const ctrl = new StreamController<StreamPoint>({
    source: (o) => makeStreamingPoints({ total, ...o }),
    onBatch: (batch) => {
      for (const f of batch) {
        f.properties.color = currentColor; // new points get the current color
        retained.push(f); // retain for repaint via the callback — loop (not spread) for large batches
      }
      const t0 = performance.now();
      points.append(batch); // incremental draw — only the new batch projects immediately
      appendMs += performance.now() - t0;
      count += batch.length;
      stats.update(count, total, count / (appendMs / 1000), ctrl.batchSize);
    },
    onReset: () => {
      retained.length = 0;
      count = 0;
      appendMs = 0;
      // Re-register the layer (clears the pass-through buffer); callback still points at retained
      points = map.layer("points", () => retained, { passThrough: true, ...pointOpts });
      stats.update(0, total, 0, ctrl.batchSize, true);
    },
  });
  // "adaptive" auto-tunes the batch to a frame budget; a number fixes it.
  const applyBatch = (opt: string): void => {
    ctrl.adaptive = opt === "adaptive";
    ctrl.batchSize = ctrl.adaptive ? ADAPTIVE_SEED_BATCH : Number(opt);
  };
  // Stream runs only when the user wants it AND the canvas is on-screen (the harness
  // calls setVisible). A manual pause persists across scroll-out/in.
  let visible = true;
  let userRunning = options.stream === "run";
  let lastBatchOpt = String(options.batch);
  applyBatch(lastBatchOpt);
  ctrl.delayMs = Number(options.rate);
  ctrl.setRunning(userRunning && visible);
  ctrl.restart(seed);

  let lastRandomize = Number(options.randomize) || 0;
  let lastRestart = Number(options.restart) || 0;

  return {
    engine: map,
    render: (o) => {
      userRunning = o.stream === "run";
      ctrl.setRunning(userRunning && visible);
      if (o.randomize !== lastRandomize) {
        lastRandomize = Number(o.randomize) || 0;
        currentColor = randomHsl(); // future points...
        for (const f of retained) f.properties.color = currentColor; // ...and already-stored ones
        points.recolor(); // re-render from the retained, updated properties
      }
      const batchOpt = String(o.batch);
      const rate = Number(o.rate);
      const newTotal = DATA_SIZE_TOTALS[String(o.size)] ?? total;
      if (
        batchOpt !== lastBatchOpt ||
        rate !== ctrl.delayMs ||
        newTotal !== total ||
        o.restart !== lastRestart
      ) {
        lastRestart = Number(o.restart) || 0;
        lastBatchOpt = batchOpt;
        applyBatch(batchOpt);
        ctrl.delayMs = rate;
        total = newTotal;
        ctrl.restart(++seed);
      }
    },
    setVisible: (v) => {
      visible = v;
      ctrl.setRunning(userRunning && visible); // pause offscreen; resume only if user wants run
    },
    dispose: () => {
      ctrl.dispose();
      stats.destroy();
    },
  };
};

geo-data.ts

geo-data.ts

import { geoGraticule } from "d3-geo";
import type { Feature, FeatureCollection, LineString, MultiLineString, MultiPoint, MultiPolygon, Point, Polygon } from "geojson";
import { feature } from "topojson-client";
import land110m from "world-atlas/land-110m.json";

/** A synthetic grid cell with a continuous value and a categorical bioregion. */
export interface Cell {
  id: string;
  geometry: Polygon;
  /** Cell centroid [lon, lat]. */
  center: [number, number];
  /** Continuous field in [0, 1] (value field). */
  value: number;
  /** Categorical bioregion id in 0..7. */
  bioregion: number;
}

/** Base cell size in degrees; the example scales it by powers of two via a slider. */
export const BASE_STEP = 1;

function clamp01(x: number): number {
  return Math.max(0, Math.min(1, x));
}

/** Generate a global grid of `step`°×`step`° cells with smooth synthetic fields. */
export function makeCells(step: number = BASE_STEP): Cell[] {
  const cells: Cell[] = [];
  let col = 0;
  for (let lon = -180; lon < 180; lon += step, col++) {
    let row = 0;
    for (let lat = -90; lat < 90; lat += step, row++) {
      const lonR = (lon * Math.PI) / 180;
      const latR = (lat * Math.PI) / 180;
      const value = clamp01(0.5 + 0.5 * Math.sin(lonR * 2) * Math.cos(latR * 3));
      const field = (Math.sin(lon / 40) + Math.cos(lat / 30)) * 0.5 + 1; // ~[0,2]
      const bioregion = Math.min(7, Math.max(0, Math.floor((field / 2) * 8)));
      const geometry: Polygon = {
        type: "Polygon",
        coordinates: [
          [
            [lon, lat],
            [lon, lat + step],
            [lon + step, lat + step],
            [lon + step, lat],
            [lon, lat],
          ],
        ],
      };
      cells.push({
        id: `${col}-${row}`,
        geometry,
        center: [lon + step / 2, lat + step / 2],
        value,
        bioregion,
      });
    }
  }
  return cells;
}

/** A fine grid over the central third of the globe (lon ±60°, lat ±30°), 4° cells —
 *  a "dense" demo layer (used clipped to land). */
export function centreCells(): Cell[] {
  return makeCells(4).filter((c) => Math.abs(c.center[0]) <= 60 && Math.abs(c.center[1]) <= 30);
}

/** Wrap cells as a FeatureCollection for projection fitting. */
export function cellsToFeatureCollection(cells: readonly Cell[]): FeatureCollection {
  const features: Feature[] = cells.map((c) => ({
    type: "Feature",
    properties: { id: c.id },
    geometry: c.geometry,
  }));
  return { type: "FeatureCollection", features };
}

/** A GeoJSON object d3-geo can fill that isn't part of the strict GeoJSON spec. */
export type Sphere = { type: "Sphere" };

/** The land outline (Natural Earth 110m) plus a sphere to fill as ocean. */
export interface World {
  sphere: Sphere;
  land: MultiPolygon;
}

// Derive the topojson Topology type from feature()'s own signature so we don't
// take a direct dependency on the (transitive) topojson-specification types.
type Topology = Parameters<typeof feature>[0];

/**
 * Convert the bundled world-atlas TopoJSON into a land MultiPolygon and a sphere.
 */
export function loadWorld(): World {
  const topo = land110m as unknown as Topology;
  const fc = feature(topo, topo.objects.land!) as unknown as FeatureCollection<MultiPolygon>;
  return { sphere: { type: "Sphere" }, land: fc.features[0]!.geometry };
}

/** A few well-known cities to show point geometry rendered alongside the grid. */
export interface City {
  id: string;
  name: string;
  geometry: Point;
}

export function makeCities(): City[] {
  const places: [string, number, number][] = [
    ["London", -0.13, 51.51],
    ["New York", -74.01, 40.71],
    ["Tokyo", 139.69, 35.69],
    ["Sydney", 151.21, -33.87],
    ["Cape Town", 18.42, -33.92],
    ["Rio de Janeiro", -43.2, -22.91],
    ["Nairobi", 36.82, -1.29],
    ["Mumbai", 72.88, 19.08],
  ];
  return places.map(([name, lon, lat]) => ({
    id: name!,
    name: name!,
    geometry: { type: "Point", coordinates: [lon!, lat!] },
  }));
}

/** A 20° graticule as one MultiLineString feature. */
export function makeGraticule(): Feature<MultiLineString> {
  return { type: "Feature", properties: {}, geometry: geoGraticule().step([20, 20])() };
}

/** A great-circle-ish route as a LineString feature (London -> New York -> Tokyo). */
export function makeRoute(): Feature<LineString> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "LineString", coordinates: [[-0.13, 51.51], [-74.01, 40.71], [139.69, 35.69]] },
  };
}

/** A cluster of locations as one MultiPoint feature. */
export function makeCluster(): Feature<MultiPoint> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "MultiPoint", coordinates: [[18.42, -33.92], [151.21, -33.87], [-43.2, -22.91], [36.82, -1.29], [72.88, 19.08]] },
  };
}

/** A standalone Polygon feature (a box over the Sahara) to showcase polygon geometry.
 *  Wound CLOCKWISE so d3-geo fills the small box (not the sphere complement). */
export function makeDemoPolygon(): Feature<Polygon> {
  return {
    type: "Feature", properties: { name: "demo-region" },
    geometry: { type: "Polygon", coordinates: [[[0, 15], [0, 30], [30, 30], [30, 15], [0, 15]]] },
  };
}

/** A handful of major rivers as rough named polylines (the bundled world-atlas data
 *  has no rivers), shown on the GeoJSON-features map and used as streaming cluster
 *  centers. Coordinates are approximate [lon, lat] traces, mouth → source. */
export function makeMajorRivers(): Feature<LineString, { name: string }>[] {
  const rivers: [string, [number, number][]][] = [
    ["Amazon", [[-50.0, -0.7], [-55.5, -2.5], [-60.0, -3.1], [-67.9, -3.5], [-73.2, -4.5]]],
    ["Nile", [[31.3, 31.4], [32.9, 24.1], [32.5, 15.6], [32.5, 9.5], [31.6, 2.3]]],
    ["Mississippi", [[-89.2, 29.2], [-90.1, 32.3], [-90.2, 38.6], [-91.2, 43.5], [-95.0, 47.2]]],
    ["Yangtze", [[121.8, 31.4], [114.3, 30.6], [106.5, 29.6], [100.2, 26.9], [94.7, 33.5]]],
    ["Congo", [[12.4, -6.0], [16.2, -4.3], [20.0, -1.0], [25.2, 0.5], [27.2, 3.0]]],
    ["Volga", [[48.0, 46.3], [45.0, 48.7], [44.5, 51.6], [47.5, 54.3], [37.0, 57.3]]],
    ["Ganges", [[90.5, 22.5], [88.0, 24.5], [83.0, 25.4], [78.0, 26.5], [78.9, 30.1]]],
  ];
  return rivers.map(([name, coordinates]) => ({
    type: "Feature",
    properties: { name },
    geometry: { type: "LineString", coordinates },
  }));
}

// ---------------------------------------------------------------------------
// Streaming sources — async generators that emit batches of features lazily
// (only `batchSize` are materialized per tick, never the whole `total`), so a
// consumer can `await`-iterate and append them live. Points/cells are CLUSTERED
// around cities + major-river vertices (not uniform), which the world-map
// examples then clip to land. Used by the "streaming data" examples.
// ---------------------------------------------------------------------------

/** A solid default color all streamed features start with (the example's
 *  "randomize" button swaps in a new color for new + retained features). */
export const DEFAULT_STREAM_COLOR = "#e23b2f";

/** Per-feature properties: a stable id (continues across batches) + a color the
 *  example owns (constant by default; swapped by the randomize button). */
export interface StreamProps {
  id: number;
  color: string;
}
export type StreamPoint = Feature<Point, StreamProps>;
export type StreamPolygon = Feature<Polygon, StreamProps>;

export interface StreamOptions {
  /** Total features emitted before the generator completes. Default 10,000,000. */
  total?: number;
  /** Features per yielded batch. A function is re-read every batch, so the caller can
   *  resize adaptively (e.g. to fit a frame budget). Default 1000. */
  batchSize?: number | (() => number);
  /** Artificial delay between batches (ms), to mirror loading from a file/network.
   *  Even 0 yields a macrotask so the browser can paint between batches. Default 0. */
  delayMs?: number;
  /** Seed for the deterministic PRNG (reproducible streams). Default 1. */
  seed?: number;
  /** Cooperative cancellation: iteration stops once `signal.aborted` is true. */
  signal?: { aborted: boolean };
}

/** Small, fast, seedable PRNG (mulberry32) — deterministic so streams reproduce. */
function mulberry32(seed: number): () => number {
  let a = seed >>> 0;
  return () => {
    a |= 0;
    a = (a + 0x6d2b79f5) | 0;
    let t = Math.imul(a ^ (a >>> 15), 1 | a);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
  };
}

const clamp = (x: number, lo: number, hi: number): number => Math.max(lo, Math.min(hi, x));

/**
 * A clustered point process (Thomas-like): many weighted "hotspot" parents at a
 * range of scales — tight, dense cities; medium clumps along rivers; and a field
 * of random global hotspots with varied spread/weight. Offspring fall around a
 * parent with an exponential (heavy-tailed) radius, so the result is patchy and
 * multi-scale — far closer to real species-occurrence density than a smooth
 * gaussian. Clip-to-land then carves the continental outline.
 */
interface Parent {
  lon: number;
  lat: number;
  /** Mean offspring radius in degrees. */
  spread: number;
  /** Relative share of points drawn from this parent. */
  weight: number;
}

function buildParents(rng: () => number): Parent[] {
  const parents: Parent[] = [];
  for (const c of makeCities())
    parents.push({ lon: c.geometry.coordinates[0]!, lat: c.geometry.coordinates[1]!, spread: 1.5, weight: 3 });
  for (const river of makeMajorRivers())
    for (const p of river.geometry.coordinates) parents.push({ lon: p[0]!, lat: p[1]!, spread: 2.5, weight: 2 });
  // Random global hotspots: rng()*rng() biases toward small spreads/weights, so
  // most clumps are tight with a few diffuse ones — a wide range of scales.
  for (let i = 0; i < 240; i++) {
    parents.push({
      lon: rng() * 360 - 180,
      lat: rng() * 160 - 80,
      spread: 0.6 + 9 * rng() * rng(),
      weight: 0.2 + 3 * rng() * rng(),
    });
  }
  return parents;
}

function cumulativeWeights(parents: readonly Parent[]): number[] {
  const cum: number[] = [];
  let s = 0;
  for (const p of parents) {
    s += p.weight;
    cum.push(s);
  }
  return cum;
}

/** Pick a parent by weight (binary search over cumulative weights). */
function pickParent(rng: () => number, parents: readonly Parent[], cum: readonly number[]): Parent {
  const r = rng() * cum[cum.length - 1]!;
  let lo = 0;
  let hi = cum.length - 1;
  while (lo < hi) {
    const mid = (lo + hi) >> 1;
    if (cum[mid]! < r) lo = mid + 1;
    else hi = mid;
  }
  return parents[lo]!;
}

/** Offspring location around a parent: exponential radius, uniform direction. */
function clusteredLonLat(rng: () => number, p: Parent): [number, number] {
  const radius = -p.spread * Math.log(Math.max(1e-9, rng())); // exponential, mean = spread
  const ang = rng() * 2 * Math.PI;
  return [
    clamp(p.lon + radius * Math.cos(ang), -180, 180),
    clamp(p.lat + radius * Math.sin(ang), -90, 90),
  ];
}

/**
 * An irregular, star-convex polygon ring (3–10 vertices, varied per-vertex radius)
 * centered at [clon, clat] with overall extent ≤ ~`size`° — a rough species range.
 * Angles are evenly spaced with bounded jitter so they stay monotonic ⇒ the ring is
 * simple (non-self-intersecting) and closed. Longitude offsets are widened by
 * 1/cos(lat) so ranges don't look squished toward the poles.
 *
 * WINDING: built CLOCKWISE in [lon, lat] (note the NEGATIVE angle). d3-geo fills on
 * the sphere, and a small exterior ring wound counter-clockwise is treated as its
 * complement → it fills the whole map. See `AGENTS.md` and `geo/project.ts`.
 */
function randomRangeRing(rng: () => number, clon: number, clat: number, size: number): [number, number][] {
  const verts = 3 + Math.floor(rng() * 8); // 3..10
  // Strongly heavy-tailed size: the vast majority of ranges are TINY and only ~3% are
  // visibly large. At high counts the translucent fill then reads as a density gradient
  // (clustered richness hotspots) instead of saturating the whole map red.
  const base =
    rng() < 0.03
      ? size * (0.1 + 0.15 * rng()) // ~3% larger ranges: 0.10..0.25 * size
      : size * (0.02 + 0.07 * rng() * rng()); // most tiny: 0.02..0.09 * size, biased small (rng²)
  const latScale = 1 / Math.max(0.25, Math.cos((clat * Math.PI) / 180));
  const ring: [number, number][] = [];
  for (let i = 0; i < verts; i++) {
    const ang = -((i + 0.5 * rng()) / verts) * 2 * Math.PI; // NEGATIVE ⇒ clockwise ⇒ fills interior
    const r = base * (0.5 + rng()); // per-vertex radius variation
    ring.push([
      clamp(clon + r * Math.cos(ang) * latScale, -180, 180),
      clamp(clat + r * Math.sin(ang), -90, 90),
    ]);
  }
  ring.push(ring[0]!); // close the ring
  return ring;
}

const tick = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));

/** Stream world points as GeoJSON `Feature<Point>` batches, clustered around
 *  many multi-scale hotspots. All start with DEFAULT_STREAM_COLOR. */
export async function* makeStreamingPoints(opts: StreamOptions = {}): AsyncGenerator<StreamPoint[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPoint[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [lon, lat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Point", coordinates: [lon, lat] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

/** Stream irregular polygon "ranges" clustered around hotspots — each a 3–10
 *  vertex star-convex polygon of varied size ≤ ~`size`°, to mimic species ranges.
 *  All start with DEFAULT_STREAM_COLOR; the example renders them very transparent
 *  so overlapping ranges build up richness. */
export async function* makeStreamingPolygons(
  opts: StreamOptions & { size?: number } = {},
): AsyncGenerator<StreamPolygon[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, size = 16, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPolygon[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [clon, clat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Polygon", coordinates: [randomRangeRing(rng, clon, clat, size)] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

streaming.ts

streaming.ts

/**
 * Drives a streaming source (an async batch generator) into a consumer, with
 * run/pause and restart — the plumbing shared by the streaming examples so each
 * `draw.ts` only has to say HOW to append a batch (the d3gl-specific part).
 */

export interface StreamSourceOpts {
  /** Re-read every batch (the controller may resize it adaptively). */
  batchSize: () => number;
  delayMs: number;
  seed: number;
  signal: { aborted: boolean };
}

export interface StreamControllerOpts<T> {
  /** Build a fresh source generator for a (re)started session. */
  source: (o: StreamSourceOpts) => AsyncGenerator<T[]>;
  /** Consume one freshly-arrived batch (append it to the layer + retain it). */
  onBatch: (batch: T[]) => void;
  /** Clear retained state + the layer at the start of a (re)started session. */
  onReset: () => void;
}

/**
 * One streaming session at a time. `restart(seed)` resets and starts a new
 * session (superseding any running one); `setRunning(false)` pauses the pump;
 * `dispose()` stops everything for good.
 *
 * Adaptive mode (`adaptive = true`): after each batch the controller times how long
 * `onBatch` (build + append + paint) took and nudges `batchSize` toward the number
 * of features that fit `frameBudgetMs` — so the stream runs as fast as the backend
 * allows while staying responsive (the old Bioregions "fit ~30fps" trick). The source
 * re-reads `batchSize` every batch via the getter passed in `restart`.
 */
export class StreamController<T> {
  batchSize = 1000;
  delayMs = 0;
  /** When true, `batchSize` is auto-tuned each batch to fit `frameBudgetMs`. */
  adaptive = false;
  /** Per-batch time budget for adaptive mode (≈30fps). */
  frameBudgetMs = 32;
  private session = 0;
  private running = true;
  private disposed = false;

  constructor(private readonly opts: StreamControllerOpts<T>) {}

  setRunning(v: boolean): void {
    this.running = v;
  }

  /** Reset + start a fresh session (supersedes any in flight). */
  restart(seed: number): void {
    if (this.disposed) return;
    this.opts.onReset();
    void this.pump(seed);
  }

  /** Stop the current session permanently (call from the example's dispose). */
  dispose(): void {
    this.disposed = true;
    this.session++;
  }

  private async pump(seed: number): Promise<void> {
    const my = ++this.session;
    const signal = { aborted: false };
    // Getter so adaptive resizes take effect on the very next batch the source pulls.
    const gen = this.opts.source({ batchSize: () => this.batchSize, delayMs: this.delayMs, seed, signal });
    for await (const batch of gen) {
      // Superseded (restart) or disposed → abort this session.
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      // Paused → idle without consuming, until resumed / superseded / disposed.
      while (!this.running && !this.disposed && my === this.session) {
        await new Promise((r) => setTimeout(r, 60));
      }
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      const t0 = performance.now();
      this.opts.onBatch(batch);
      if (this.adaptive) {
        // Steer toward the budget: scale by budget/elapsed, clamped to ≤2× per step so
        // it ramps quickly but doesn't oscillate. Bounded to [1, 1_000_000].
        const dt = Math.max(0.5, performance.now() - t0);
        const factor = Math.max(0.5, Math.min(2, this.frameBudgetMs / dt));
        this.batchSize = Math.max(1, Math.min(1_000_000, Math.round(this.batchSize * factor)));
      }
    }
  }
}

/** A random vivid color (used by the "randomize colors" button). Pass `alpha` < 1
 *  for translucent fills (e.g. overlapping polygon ranges). */
export function randomHsl(alpha = 1): string {
  const h = Math.floor(Math.random() * 360);
  return alpha >= 1 ? `hsl(${h}, 70%, 55%)` : `hsla(${h}, 70%, 55%, ${alpha})`;
}

/** The batch-size choices offered by the streaming examples' control. "adaptive"
 *  auto-tunes the batch to fit a frame budget (see StreamController.adaptive). */
export const BATCH_SIZES = ["adaptive", "1", "10", "100", "1000", "100000", "1000000"];
/** Seed batch size when adaptive mode starts, before it converges. */
export const ADAPTIVE_SEED_BATCH = 256;
/** The artificial per-batch delay choices (ms) — mirrors loading from a file. */
export const RATES_MS = ["0", "16", "100", "500"];
/** Data-size choices (total features to stream) and their numeric totals. */
export const DATA_SIZES = ["100k", "1M", "10M"];
export const DATA_SIZE_TOTALS: Record<string, number> = {
  "100k": 100_000,
  "1M": 1_000_000,
  "10M": 10_000_000,
};

stats-overlay.ts

stats-overlay.ts

/**
 * A tiny HTML overlay (top-right of the canvas) showing how many records have
 * streamed in — count, percentage of the total, and the average append speed so
 * far (records / time spent in `append`). Mirrors the Infomap Bioregions readout.
 */
export interface StatsOverlay {
  /** Update the readout. Throttled to ~10 Hz unless `force`. `batch` (optional) shows
   *  the current batch size — handy to watch adaptive batching converge. */
  update(count: number, total: number, recordsPerSec: number, batch?: number, force?: boolean): void;
  destroy(): void;
}

export function createStatsOverlay(host: HTMLElement): StatsOverlay {
  const el = document.createElement("div");
  el.className =
    "absolute top-2 right-2 pointer-events-none rounded-md bg-white/80 px-2 py-1 " +
    "font-mono text-[11px] leading-tight text-[#333] shadow-sm [font-variant-numeric:tabular-nums]";
  host.appendChild(el);

  const fmt = (n: number): string => Math.round(n).toLocaleString();
  let last = 0;

  return {
    update(count, total, recordsPerSec, batch, force = false) {
      const now = performance.now();
      if (!force && now - last < 100) return;
      last = now;
      const ratio = total > 0 ? count / total : 0;
      const pct = (ratio * 100).toFixed(ratio < 0.1 ? 1 : 0);
      const batchLine = batch != null ? `<br>batch <b>${fmt(batch)}</b>` : "";
      el.innerHTML = `records <b>${fmt(count)}</b> (${pct}%)<br>speed <b>${fmt(recordsPerSec)}</b> rec/s${batchLine}`;
    },
    destroy() {
      el.remove();
    },
  };
}

Pass-through polygons

Pass-through works for polygons and lines too, not just points. On WebGL the full set re-tessellates on every pan/zoom settle (the documented cost of pass-through for path geometry); on Canvas it re-projects paths, same as retained. The demo below streams 1M translucent polygon ranges via the pass-through path — raise “Data size” to push the ceiling further. Unlike the retained polygons example above, these ranges are not clipped to land — clipTo isn’t applied to pass-through layers yet (see the trade-off table), so ranges appear over the ocean too.

autowebglcanvassvg

Export PNG

fps 0frame 0 ms

Stream

runpause

RestartRandomize colors

draw.ts

draw.ts

import { geoEquirectangular } from "d3-geo";
import { geoMap, type LayerHandle } from "@mapequation/d3gl/map";
import { loadWorld, makeStreamingPolygons, type StreamPolygon } from "../shared/geo-data.js";
import { StreamController, randomHsl, DATA_SIZE_TOTALS, ADAPTIVE_SEED_BATCH } from "../shared/streaming.js";
import { createStatsOverlay } from "../shared/stats-overlay.js";
import type { ImperativeSetup } from "../types.js";

const OCEAN = "#dbe7f3";
const LAND = "#e9e7df";
const RANGE_ALPHA = 0.05; // very transparent so overlapping ranges build up a density gradient

/**
 * Stream polygon "ranges" using the pass-through path (`passThrough: true`).
 * The engine re-reads `() => retained` on each repaint — so pan/zoom re-projects
 * (and on WebGL re-tessellates) the full set, with no ceiling inside d3gl.
 *
 * Contrast with `streaming-polygons` (retained path): that example supports
 * picking and per-feature recolor; this one is uncapped and not pickable.
 * New batches draw immediately via `cells.append(batch)`; the callback
 * covers full repaints (pan/zoom settle).
 *
 * WebGL re-tessellates the full set on every pan/zoom settle — the documented
 * cost of pass-through for polygon/line geometry. Use a modest data-size default
 * so the demo stays responsive; raise it if you want to test limits.
 */
export const setup: ImperativeSetup = (host, { width, height, backend, options }) => {
  const world = loadWorld();
  const projection = geoEquirectangular().fitSize([width, height], { type: "Sphere" });
  const map = geoMap(host, { width, height, projection, backend });
  map.layer("ocean", [world.sphere], { fill: OCEAN });
  map.layer("land", [world.land], { fill: LAND, stroke: "#9aa3ad", lineWidth: 0.4 });
  map.enableZoom([1, 40]);

  const retained: StreamPolygon[] = [];
  let currentColor = `hsla(8, 80%, 53%, ${RANGE_ALPHA})`; // translucent red; new ranges get this
  const cellOpts = {
    fill: (f: StreamPolygon) => f.properties.color,
    // pass-through layers are not pickable (no hit index); no id/clipTo needed
  };
  // passThrough: true — engine re-invokes `() => retained` each repaint (pan/zoom)
  let cells: LayerHandle<StreamPolygon> = map.layer("cells", () => retained, {
    passThrough: true,
    ...cellOpts,
  });
  map.render();

  const stats = createStatsOverlay(host);
  let count = 0;
  let appendMs = 0;

  let seed = 1;
  let total = DATA_SIZE_TOTALS[String(options.size)] ?? 100_000;
  const ctrl = new StreamController<StreamPolygon>({
    source: (o) => makeStreamingPolygons({ total, ...o }),
    onBatch: (batch) => {
      for (const f of batch) {
        f.properties.color = currentColor; // new polygons get the current color
        retained.push(f); // retain for repaint via the callback — loop (not spread) for large batches
      }
      const t0 = performance.now();
      cells.append(batch); // incremental draw — only the new batch projects/tessellates immediately
      appendMs += performance.now() - t0;
      count += batch.length;
      stats.update(count, total, count / (appendMs / 1000), ctrl.batchSize);
    },
    onReset: () => {
      retained.length = 0;
      count = 0;
      appendMs = 0;
      // Re-register the layer (clears the pass-through buffer); callback still points at retained
      cells = map.layer("cells", () => retained, { passThrough: true, ...cellOpts });
      stats.update(0, total, 0, ctrl.batchSize, true);
    },
  });
  // "adaptive" auto-tunes the batch to a frame budget; a number fixes it.
  const applyBatch = (opt: string): void => {
    ctrl.adaptive = opt === "adaptive";
    ctrl.batchSize = ctrl.adaptive ? ADAPTIVE_SEED_BATCH : Number(opt);
  };
  // Stream runs only when the user wants it AND the canvas is on-screen (the harness
  // calls setVisible). A manual pause persists across scroll-out/in.
  let visible = true;
  let userRunning = options.stream === "run";
  let lastBatchOpt = String(options.batch);
  applyBatch(lastBatchOpt);
  ctrl.delayMs = Number(options.rate);
  ctrl.setRunning(userRunning && visible);
  ctrl.restart(seed);

  let lastRandomize = Number(options.randomize) || 0;
  let lastRestart = Number(options.restart) || 0;

  return {
    engine: map,
    render: (o) => {
      userRunning = o.stream === "run";
      ctrl.setRunning(userRunning && visible);
      if (o.randomize !== lastRandomize) {
        lastRandomize = Number(o.randomize) || 0;
        currentColor = randomHsl(RANGE_ALPHA); // keep ranges translucent
        for (const f of retained) f.properties.color = currentColor;
        cells.recolor(); // re-render from the retained, updated properties
      }
      const batchOpt = String(o.batch);
      const rate = Number(o.rate);
      const newTotal = DATA_SIZE_TOTALS[String(o.size)] ?? total;
      if (
        batchOpt !== lastBatchOpt ||
        rate !== ctrl.delayMs ||
        newTotal !== total ||
        o.restart !== lastRestart
      ) {
        lastRestart = Number(o.restart) || 0;
        lastBatchOpt = batchOpt;
        applyBatch(batchOpt);
        ctrl.delayMs = rate;
        total = newTotal;
        ctrl.restart(++seed);
      }
    },
    setVisible: (v) => {
      visible = v;
      ctrl.setRunning(userRunning && visible); // pause offscreen; resume only if user wants run
    },
    dispose: () => {
      ctrl.dispose();
      stats.destroy();
    },
  };
};

geo-data.ts

geo-data.ts

import { geoGraticule } from "d3-geo";
import type { Feature, FeatureCollection, LineString, MultiLineString, MultiPoint, MultiPolygon, Point, Polygon } from "geojson";
import { feature } from "topojson-client";
import land110m from "world-atlas/land-110m.json";

/** A synthetic grid cell with a continuous value and a categorical bioregion. */
export interface Cell {
  id: string;
  geometry: Polygon;
  /** Cell centroid [lon, lat]. */
  center: [number, number];
  /** Continuous field in [0, 1] (value field). */
  value: number;
  /** Categorical bioregion id in 0..7. */
  bioregion: number;
}

/** Base cell size in degrees; the example scales it by powers of two via a slider. */
export const BASE_STEP = 1;

function clamp01(x: number): number {
  return Math.max(0, Math.min(1, x));
}

/** Generate a global grid of `step`°×`step`° cells with smooth synthetic fields. */
export function makeCells(step: number = BASE_STEP): Cell[] {
  const cells: Cell[] = [];
  let col = 0;
  for (let lon = -180; lon < 180; lon += step, col++) {
    let row = 0;
    for (let lat = -90; lat < 90; lat += step, row++) {
      const lonR = (lon * Math.PI) / 180;
      const latR = (lat * Math.PI) / 180;
      const value = clamp01(0.5 + 0.5 * Math.sin(lonR * 2) * Math.cos(latR * 3));
      const field = (Math.sin(lon / 40) + Math.cos(lat / 30)) * 0.5 + 1; // ~[0,2]
      const bioregion = Math.min(7, Math.max(0, Math.floor((field / 2) * 8)));
      const geometry: Polygon = {
        type: "Polygon",
        coordinates: [
          [
            [lon, lat],
            [lon, lat + step],
            [lon + step, lat + step],
            [lon + step, lat],
            [lon, lat],
          ],
        ],
      };
      cells.push({
        id: `${col}-${row}`,
        geometry,
        center: [lon + step / 2, lat + step / 2],
        value,
        bioregion,
      });
    }
  }
  return cells;
}

/** A fine grid over the central third of the globe (lon ±60°, lat ±30°), 4° cells —
 *  a "dense" demo layer (used clipped to land). */
export function centreCells(): Cell[] {
  return makeCells(4).filter((c) => Math.abs(c.center[0]) <= 60 && Math.abs(c.center[1]) <= 30);
}

/** Wrap cells as a FeatureCollection for projection fitting. */
export function cellsToFeatureCollection(cells: readonly Cell[]): FeatureCollection {
  const features: Feature[] = cells.map((c) => ({
    type: "Feature",
    properties: { id: c.id },
    geometry: c.geometry,
  }));
  return { type: "FeatureCollection", features };
}

/** A GeoJSON object d3-geo can fill that isn't part of the strict GeoJSON spec. */
export type Sphere = { type: "Sphere" };

/** The land outline (Natural Earth 110m) plus a sphere to fill as ocean. */
export interface World {
  sphere: Sphere;
  land: MultiPolygon;
}

// Derive the topojson Topology type from feature()'s own signature so we don't
// take a direct dependency on the (transitive) topojson-specification types.
type Topology = Parameters<typeof feature>[0];

/**
 * Convert the bundled world-atlas TopoJSON into a land MultiPolygon and a sphere.
 */
export function loadWorld(): World {
  const topo = land110m as unknown as Topology;
  const fc = feature(topo, topo.objects.land!) as unknown as FeatureCollection<MultiPolygon>;
  return { sphere: { type: "Sphere" }, land: fc.features[0]!.geometry };
}

/** A few well-known cities to show point geometry rendered alongside the grid. */
export interface City {
  id: string;
  name: string;
  geometry: Point;
}

export function makeCities(): City[] {
  const places: [string, number, number][] = [
    ["London", -0.13, 51.51],
    ["New York", -74.01, 40.71],
    ["Tokyo", 139.69, 35.69],
    ["Sydney", 151.21, -33.87],
    ["Cape Town", 18.42, -33.92],
    ["Rio de Janeiro", -43.2, -22.91],
    ["Nairobi", 36.82, -1.29],
    ["Mumbai", 72.88, 19.08],
  ];
  return places.map(([name, lon, lat]) => ({
    id: name!,
    name: name!,
    geometry: { type: "Point", coordinates: [lon!, lat!] },
  }));
}

/** A 20° graticule as one MultiLineString feature. */
export function makeGraticule(): Feature<MultiLineString> {
  return { type: "Feature", properties: {}, geometry: geoGraticule().step([20, 20])() };
}

/** A great-circle-ish route as a LineString feature (London -> New York -> Tokyo). */
export function makeRoute(): Feature<LineString> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "LineString", coordinates: [[-0.13, 51.51], [-74.01, 40.71], [139.69, 35.69]] },
  };
}

/** A cluster of locations as one MultiPoint feature. */
export function makeCluster(): Feature<MultiPoint> {
  return {
    type: "Feature", properties: {},
    geometry: { type: "MultiPoint", coordinates: [[18.42, -33.92], [151.21, -33.87], [-43.2, -22.91], [36.82, -1.29], [72.88, 19.08]] },
  };
}

/** A standalone Polygon feature (a box over the Sahara) to showcase polygon geometry.
 *  Wound CLOCKWISE so d3-geo fills the small box (not the sphere complement). */
export function makeDemoPolygon(): Feature<Polygon> {
  return {
    type: "Feature", properties: { name: "demo-region" },
    geometry: { type: "Polygon", coordinates: [[[0, 15], [0, 30], [30, 30], [30, 15], [0, 15]]] },
  };
}

/** A handful of major rivers as rough named polylines (the bundled world-atlas data
 *  has no rivers), shown on the GeoJSON-features map and used as streaming cluster
 *  centers. Coordinates are approximate [lon, lat] traces, mouth → source. */
export function makeMajorRivers(): Feature<LineString, { name: string }>[] {
  const rivers: [string, [number, number][]][] = [
    ["Amazon", [[-50.0, -0.7], [-55.5, -2.5], [-60.0, -3.1], [-67.9, -3.5], [-73.2, -4.5]]],
    ["Nile", [[31.3, 31.4], [32.9, 24.1], [32.5, 15.6], [32.5, 9.5], [31.6, 2.3]]],
    ["Mississippi", [[-89.2, 29.2], [-90.1, 32.3], [-90.2, 38.6], [-91.2, 43.5], [-95.0, 47.2]]],
    ["Yangtze", [[121.8, 31.4], [114.3, 30.6], [106.5, 29.6], [100.2, 26.9], [94.7, 33.5]]],
    ["Congo", [[12.4, -6.0], [16.2, -4.3], [20.0, -1.0], [25.2, 0.5], [27.2, 3.0]]],
    ["Volga", [[48.0, 46.3], [45.0, 48.7], [44.5, 51.6], [47.5, 54.3], [37.0, 57.3]]],
    ["Ganges", [[90.5, 22.5], [88.0, 24.5], [83.0, 25.4], [78.0, 26.5], [78.9, 30.1]]],
  ];
  return rivers.map(([name, coordinates]) => ({
    type: "Feature",
    properties: { name },
    geometry: { type: "LineString", coordinates },
  }));
}

// ---------------------------------------------------------------------------
// Streaming sources — async generators that emit batches of features lazily
// (only `batchSize` are materialized per tick, never the whole `total`), so a
// consumer can `await`-iterate and append them live. Points/cells are CLUSTERED
// around cities + major-river vertices (not uniform), which the world-map
// examples then clip to land. Used by the "streaming data" examples.
// ---------------------------------------------------------------------------

/** A solid default color all streamed features start with (the example's
 *  "randomize" button swaps in a new color for new + retained features). */
export const DEFAULT_STREAM_COLOR = "#e23b2f";

/** Per-feature properties: a stable id (continues across batches) + a color the
 *  example owns (constant by default; swapped by the randomize button). */
export interface StreamProps {
  id: number;
  color: string;
}
export type StreamPoint = Feature<Point, StreamProps>;
export type StreamPolygon = Feature<Polygon, StreamProps>;

export interface StreamOptions {
  /** Total features emitted before the generator completes. Default 10,000,000. */
  total?: number;
  /** Features per yielded batch. A function is re-read every batch, so the caller can
   *  resize adaptively (e.g. to fit a frame budget). Default 1000. */
  batchSize?: number | (() => number);
  /** Artificial delay between batches (ms), to mirror loading from a file/network.
   *  Even 0 yields a macrotask so the browser can paint between batches. Default 0. */
  delayMs?: number;
  /** Seed for the deterministic PRNG (reproducible streams). Default 1. */
  seed?: number;
  /** Cooperative cancellation: iteration stops once `signal.aborted` is true. */
  signal?: { aborted: boolean };
}

/** Small, fast, seedable PRNG (mulberry32) — deterministic so streams reproduce. */
function mulberry32(seed: number): () => number {
  let a = seed >>> 0;
  return () => {
    a |= 0;
    a = (a + 0x6d2b79f5) | 0;
    let t = Math.imul(a ^ (a >>> 15), 1 | a);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
  };
}

const clamp = (x: number, lo: number, hi: number): number => Math.max(lo, Math.min(hi, x));

/**
 * A clustered point process (Thomas-like): many weighted "hotspot" parents at a
 * range of scales — tight, dense cities; medium clumps along rivers; and a field
 * of random global hotspots with varied spread/weight. Offspring fall around a
 * parent with an exponential (heavy-tailed) radius, so the result is patchy and
 * multi-scale — far closer to real species-occurrence density than a smooth
 * gaussian. Clip-to-land then carves the continental outline.
 */
interface Parent {
  lon: number;
  lat: number;
  /** Mean offspring radius in degrees. */
  spread: number;
  /** Relative share of points drawn from this parent. */
  weight: number;
}

function buildParents(rng: () => number): Parent[] {
  const parents: Parent[] = [];
  for (const c of makeCities())
    parents.push({ lon: c.geometry.coordinates[0]!, lat: c.geometry.coordinates[1]!, spread: 1.5, weight: 3 });
  for (const river of makeMajorRivers())
    for (const p of river.geometry.coordinates) parents.push({ lon: p[0]!, lat: p[1]!, spread: 2.5, weight: 2 });
  // Random global hotspots: rng()*rng() biases toward small spreads/weights, so
  // most clumps are tight with a few diffuse ones — a wide range of scales.
  for (let i = 0; i < 240; i++) {
    parents.push({
      lon: rng() * 360 - 180,
      lat: rng() * 160 - 80,
      spread: 0.6 + 9 * rng() * rng(),
      weight: 0.2 + 3 * rng() * rng(),
    });
  }
  return parents;
}

function cumulativeWeights(parents: readonly Parent[]): number[] {
  const cum: number[] = [];
  let s = 0;
  for (const p of parents) {
    s += p.weight;
    cum.push(s);
  }
  return cum;
}

/** Pick a parent by weight (binary search over cumulative weights). */
function pickParent(rng: () => number, parents: readonly Parent[], cum: readonly number[]): Parent {
  const r = rng() * cum[cum.length - 1]!;
  let lo = 0;
  let hi = cum.length - 1;
  while (lo < hi) {
    const mid = (lo + hi) >> 1;
    if (cum[mid]! < r) lo = mid + 1;
    else hi = mid;
  }
  return parents[lo]!;
}

/** Offspring location around a parent: exponential radius, uniform direction. */
function clusteredLonLat(rng: () => number, p: Parent): [number, number] {
  const radius = -p.spread * Math.log(Math.max(1e-9, rng())); // exponential, mean = spread
  const ang = rng() * 2 * Math.PI;
  return [
    clamp(p.lon + radius * Math.cos(ang), -180, 180),
    clamp(p.lat + radius * Math.sin(ang), -90, 90),
  ];
}

/**
 * An irregular, star-convex polygon ring (3–10 vertices, varied per-vertex radius)
 * centered at [clon, clat] with overall extent ≤ ~`size`° — a rough species range.
 * Angles are evenly spaced with bounded jitter so they stay monotonic ⇒ the ring is
 * simple (non-self-intersecting) and closed. Longitude offsets are widened by
 * 1/cos(lat) so ranges don't look squished toward the poles.
 *
 * WINDING: built CLOCKWISE in [lon, lat] (note the NEGATIVE angle). d3-geo fills on
 * the sphere, and a small exterior ring wound counter-clockwise is treated as its
 * complement → it fills the whole map. See `AGENTS.md` and `geo/project.ts`.
 */
function randomRangeRing(rng: () => number, clon: number, clat: number, size: number): [number, number][] {
  const verts = 3 + Math.floor(rng() * 8); // 3..10
  // Strongly heavy-tailed size: the vast majority of ranges are TINY and only ~3% are
  // visibly large. At high counts the translucent fill then reads as a density gradient
  // (clustered richness hotspots) instead of saturating the whole map red.
  const base =
    rng() < 0.03
      ? size * (0.1 + 0.15 * rng()) // ~3% larger ranges: 0.10..0.25 * size
      : size * (0.02 + 0.07 * rng() * rng()); // most tiny: 0.02..0.09 * size, biased small (rng²)
  const latScale = 1 / Math.max(0.25, Math.cos((clat * Math.PI) / 180));
  const ring: [number, number][] = [];
  for (let i = 0; i < verts; i++) {
    const ang = -((i + 0.5 * rng()) / verts) * 2 * Math.PI; // NEGATIVE ⇒ clockwise ⇒ fills interior
    const r = base * (0.5 + rng()); // per-vertex radius variation
    ring.push([
      clamp(clon + r * Math.cos(ang) * latScale, -180, 180),
      clamp(clat + r * Math.sin(ang), -90, 90),
    ]);
  }
  ring.push(ring[0]!); // close the ring
  return ring;
}

const tick = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));

/** Stream world points as GeoJSON `Feature<Point>` batches, clustered around
 *  many multi-scale hotspots. All start with DEFAULT_STREAM_COLOR. */
export async function* makeStreamingPoints(opts: StreamOptions = {}): AsyncGenerator<StreamPoint[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPoint[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [lon, lat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Point", coordinates: [lon, lat] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

/** Stream irregular polygon "ranges" clustered around hotspots — each a 3–10
 *  vertex star-convex polygon of varied size ≤ ~`size`°, to mimic species ranges.
 *  All start with DEFAULT_STREAM_COLOR; the example renders them very transparent
 *  so overlapping ranges build up richness. */
export async function* makeStreamingPolygons(
  opts: StreamOptions & { size?: number } = {},
): AsyncGenerator<StreamPolygon[]> {
  const { total = 10_000_000, batchSize = 1000, delayMs = 0, seed = 1, size = 16, signal } = opts;
  const sizeOf = (): number => Math.max(1, Math.floor(typeof batchSize === "function" ? batchSize() : batchSize));
  const rng = mulberry32(seed);
  const parents = buildParents(rng);
  const cum = cumulativeWeights(parents);
  let id = 0;
  while (id < total) {
    if (signal?.aborted) return;
    const n = Math.min(sizeOf(), total - id);
    const batch: StreamPolygon[] = new Array(n);
    for (let k = 0; k < n; k++) {
      const [clon, clat] = clusteredLonLat(rng, pickParent(rng, parents, cum));
      batch[k] = {
        type: "Feature",
        properties: { id: id++, color: DEFAULT_STREAM_COLOR },
        geometry: { type: "Polygon", coordinates: [randomRangeRing(rng, clon, clat, size)] },
      };
    }
    yield batch;
    await tick(delayMs);
  }
}

streaming.ts

streaming.ts

/**
 * Drives a streaming source (an async batch generator) into a consumer, with
 * run/pause and restart — the plumbing shared by the streaming examples so each
 * `draw.ts` only has to say HOW to append a batch (the d3gl-specific part).
 */

export interface StreamSourceOpts {
  /** Re-read every batch (the controller may resize it adaptively). */
  batchSize: () => number;
  delayMs: number;
  seed: number;
  signal: { aborted: boolean };
}

export interface StreamControllerOpts<T> {
  /** Build a fresh source generator for a (re)started session. */
  source: (o: StreamSourceOpts) => AsyncGenerator<T[]>;
  /** Consume one freshly-arrived batch (append it to the layer + retain it). */
  onBatch: (batch: T[]) => void;
  /** Clear retained state + the layer at the start of a (re)started session. */
  onReset: () => void;
}

/**
 * One streaming session at a time. `restart(seed)` resets and starts a new
 * session (superseding any running one); `setRunning(false)` pauses the pump;
 * `dispose()` stops everything for good.
 *
 * Adaptive mode (`adaptive = true`): after each batch the controller times how long
 * `onBatch` (build + append + paint) took and nudges `batchSize` toward the number
 * of features that fit `frameBudgetMs` — so the stream runs as fast as the backend
 * allows while staying responsive (the old Bioregions "fit ~30fps" trick). The source
 * re-reads `batchSize` every batch via the getter passed in `restart`.
 */
export class StreamController<T> {
  batchSize = 1000;
  delayMs = 0;
  /** When true, `batchSize` is auto-tuned each batch to fit `frameBudgetMs`. */
  adaptive = false;
  /** Per-batch time budget for adaptive mode (≈30fps). */
  frameBudgetMs = 32;
  private session = 0;
  private running = true;
  private disposed = false;

  constructor(private readonly opts: StreamControllerOpts<T>) {}

  setRunning(v: boolean): void {
    this.running = v;
  }

  /** Reset + start a fresh session (supersedes any in flight). */
  restart(seed: number): void {
    if (this.disposed) return;
    this.opts.onReset();
    void this.pump(seed);
  }

  /** Stop the current session permanently (call from the example's dispose). */
  dispose(): void {
    this.disposed = true;
    this.session++;
  }

  private async pump(seed: number): Promise<void> {
    const my = ++this.session;
    const signal = { aborted: false };
    // Getter so adaptive resizes take effect on the very next batch the source pulls.
    const gen = this.opts.source({ batchSize: () => this.batchSize, delayMs: this.delayMs, seed, signal });
    for await (const batch of gen) {
      // Superseded (restart) or disposed → abort this session.
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      // Paused → idle without consuming, until resumed / superseded / disposed.
      while (!this.running && !this.disposed && my === this.session) {
        await new Promise((r) => setTimeout(r, 60));
      }
      if (this.disposed || my !== this.session) {
        signal.aborted = true;
        return;
      }
      const t0 = performance.now();
      this.opts.onBatch(batch);
      if (this.adaptive) {
        // Steer toward the budget: scale by budget/elapsed, clamped to ≤2× per step so
        // it ramps quickly but doesn't oscillate. Bounded to [1, 1_000_000].
        const dt = Math.max(0.5, performance.now() - t0);
        const factor = Math.max(0.5, Math.min(2, this.frameBudgetMs / dt));
        this.batchSize = Math.max(1, Math.min(1_000_000, Math.round(this.batchSize * factor)));
      }
    }
  }
}

/** A random vivid color (used by the "randomize colors" button). Pass `alpha` < 1
 *  for translucent fills (e.g. overlapping polygon ranges). */
export function randomHsl(alpha = 1): string {
  const h = Math.floor(Math.random() * 360);
  return alpha >= 1 ? `hsl(${h}, 70%, 55%)` : `hsla(${h}, 70%, 55%, ${alpha})`;
}

/** The batch-size choices offered by the streaming examples' control. "adaptive"
 *  auto-tunes the batch to fit a frame budget (see StreamController.adaptive). */
export const BATCH_SIZES = ["adaptive", "1", "10", "100", "1000", "100000", "1000000"];
/** Seed batch size when adaptive mode starts, before it converges. */
export const ADAPTIVE_SEED_BATCH = 256;
/** The artificial per-batch delay choices (ms) — mirrors loading from a file. */
export const RATES_MS = ["0", "16", "100", "500"];
/** Data-size choices (total features to stream) and their numeric totals. */
export const DATA_SIZES = ["100k", "1M", "10M"];
export const DATA_SIZE_TOTALS: Record<string, number> = {
  "100k": 100_000,
  "1M": 1_000_000,
  "10M": 10_000_000,
};

stats-overlay.ts

stats-overlay.ts

/**
 * A tiny HTML overlay (top-right of the canvas) showing how many records have
 * streamed in — count, percentage of the total, and the average append speed so
 * far (records / time spent in `append`). Mirrors the Infomap Bioregions readout.
 */
export interface StatsOverlay {
  /** Update the readout. Throttled to ~10 Hz unless `force`. `batch` (optional) shows
   *  the current batch size — handy to watch adaptive batching converge. */
  update(count: number, total: number, recordsPerSec: number, batch?: number, force?: boolean): void;
  destroy(): void;
}

export function createStatsOverlay(host: HTMLElement): StatsOverlay {
  const el = document.createElement("div");
  el.className =
    "absolute top-2 right-2 pointer-events-none rounded-md bg-white/80 px-2 py-1 " +
    "font-mono text-[11px] leading-tight text-[#333] shadow-sm [font-variant-numeric:tabular-nums]";
  host.appendChild(el);

  const fmt = (n: number): string => Math.round(n).toLocaleString();
  let last = 0;

  return {
    update(count, total, recordsPerSec, batch, force = false) {
      const now = performance.now();
      if (!force && now - last < 100) return;
      last = now;
      const ratio = total > 0 ? count / total : 0;
      const pct = (ratio * 100).toFixed(ratio < 0.1 ? 1 : 0);
      const batchLine = batch != null ? `<br>batch <b>${fmt(batch)}</b>` : "";
      el.innerHTML = `records <b>${fmt(count)}</b> (${pct}%)<br>speed <b>${fmt(recordsPerSec)}</b> rec/s${batchLine}`;
    },
    destroy() {
      el.remove();
    },
  };
}