Emitter and frames

When you control both ends of a stream, skip text. An Emitter (Node only) is a TCP server that stream workers connect to, usually one connection per worker; each batch you write goes out as column frames, dealt round-robin to the connections that aren’t backed up. In Vectro frames, numbers travel as raw float64 and categories as u32 codes with each label sent once per connection, so the receiving worker copies number columns straight into memory with no parsing. With format: 'arrow' the emitter sends an Arrow IPC stream per connection instead, which pyarrow, Polars and other Arrow readers understand. BatchEncoder, FrameReader, encodeSchema and encodeEnd read and write frames without the emitter, in any runtime.

import { Emitter, BatchEncoder, FrameReader, encodeSchema, encodeEnd } from 'vectrojs';
import { BatchEncoder, FrameReader, encodeSchema, encodeEnd } from 'vectrojs/browser';   // or 'vectrojs/portable'

Emitter

Emitter.listen(options)

Starts listening and returns a promise of the Emitter. Each connection gets the schema first.

Name	Type	Default	Meaning
columns	object	required	{ name: 'number' or 'category' }, in the order the frames carry them. With format: 'arrow', also 'float32' and 'date'.
port	number	0 (any free port)	The TCP port.
host	string	'127.0.0.1'	The address to listen on.
format	'vectro' or 'arrow'	'vectro'	Vectro frames, or an Arrow IPC stream per connection.

The receiver opens its stream with the same columns and format, then connects each worker:

const feed = await Emitter.listen({ port: 9000, columns: { ts: 'number', price: 'number', region: 'category' } });
// elsewhere
const orders = await engine.streamAsync({ format: 'vectro', columns: { ts: 'number', price: 'number', region: 'category' } });
await Promise.all(engine.workers.map((_, w) => orders.connect(w, { host: 'feed.local', port: 9000 })));

feed.port

A getter: the port it listens on (useful with port: 0).

feed.connections

A getter: the number of open connections.

feed.waitForConnections(n)

Resolves once n connections are open.

feed.write(batch)

Sends one batch: { column: values }, every column the same length. Number columns are a Float64Array or number[]; category columns a string[] or, faster, { codes, labels } (codes into a labels array, which skips the per-row label lookup). Batches larger than 65,536 rows go out as several frames. Each frame goes to the next connection with room; when every connection has 8 MB unsent the promise waits (TCP backpressure), and it waits for a first connection if there is none. The rowsSent property counts rows written.

feed.close()

Ends every connection cleanly (with an end frame) and stops listening. Returns a promise.

import { Emitter, SplitEngine } from 'vectrojs';

const columns = { ts: 'number', price: 'number', region: 'category' };
const feed = await Emitter.listen({ columns });

const engine = new SplitEngine({ workers: 2 });
const orders = await engine.streamAsync({ format: 'vectro', columns });
await Promise.all(engine.workers.map((_, w) => orders.connect(w, { port: feed.port })));
await feed.waitForConnections(2);

const REGIONS = ['EU', 'US', 'APAC'];
await feed.write({ ts: [1, 2, 3], price: [10, 20, 30], region: ['EU', 'US', 'EU'] });
await feed.write({ ts: Float64Array.of(4, 5), price: Float64Array.of(40, 50), region: { codes: [2, 0], labels: REGIONS } });

while ((await orders.flushAsync()).rows < feed.rowsSent) await new Promise((r) => setTimeout(r, 10));
console.log(feed.connections, feed.rowsSent);
console.log((await orders.groupBy('region').aggAsync({ n: 'count()', revenue: 'sum(price)' })).toRows());

await feed.close();
await engine.close();

Frames

The layout, as documented at the top of src/frames.mjs. A connection carries frames, each:

Bytes	Meaning
u32	Length of the rest of the frame, little-endian.
u8	Kind: 1 schema, 2 batch, 3 end.
u8, u16	Reserved, 0.

• Schema: a u32 byte length, then UTF-8 JSON { "v": 1, "columns": [[name, kind], ...] }.
• Batch: u32 rows, u32 0, then each column in schema order, padded to 8 bytes. A number column is rows x float64. A category column is a u32 count of new labels (then a u32 0), each new label as a u32 byte length and UTF-8 bytes padded to 8, then rows x u32 codes. Codes number the labels in order of first appearance on the connection.
• End: nothing more.

Every section starts at a multiple of 8 from the frame start, so a frame copied to an aligned buffer can be read through typed-array views. Over WebSockets, send each frame as one binary message.

encodeSchema(columns)

The schema frame for { name: 'number' or 'category' }, as a Uint8Array.

encodeEnd()

The end frame, as a Uint8Array.

new BatchEncoder(columns)

Builds batch frames for one connection; it remembers which labels that connection has already received. columns as for encodeSchema.

encoder.encode(batch)

One batch frame, as bytes. batch as for feed.write, but sent as one frame whatever its size.

new FrameReader(handlers)

Reads frames from bytes that arrive in pieces cut anywhere. handlers: onSchema(columns) with { name: kind }; onBatch(rows, columns) with columns[k] = { values: Float64Array } or { codes: Uint32Array, labels } (every label of this connection so far); and optionally onEnd(). The arrays point into a buffer the reader reuses: copy what you keep before returning.

reader.push(chunk)

Feeds a Uint8Array; handlers run for every complete frame.

import { BatchEncoder, FrameReader, encodeSchema, encodeEnd } from 'vectrojs';

const columns = { price: 'number', region: 'category' };
const enc = new BatchEncoder(columns);
const frames = [
  encodeSchema(columns),
  enc.encode({ price: [10, 20], region: ['EU', 'US'] }),
  enc.encode({ price: [30], region: ['EU'] }),   // 'EU' is not sent again
  encodeEnd(),
];

const reader = new FrameReader({
  onSchema: (cols) => console.log('schema', cols),
  onBatch: (rows, [price, region]) => console.log(rows, Array.from(price.values), Array.from(region.codes, (c) => region.labels[c])),
  onEnd: () => console.log('end'),
});
for (const f of frames) for (let i = 0; i < f.length; i += 5) reader.push(f.subarray(i, i + 5));
console.log(frames.map((f) => f.length));