Streams, bytes and files

A message body can carry octets. Three kinds of value do it, and they all cross a connection the same way: as a run of bytes, sent in chunks at the pace the reader takes them.

This works in both directions, inside any plain object or array, and for the value a reflection method returns. A local fork (a child spawned in this process) hands the child the value itself; the encoding happens only when a connection is in the way.

Typed arrays

proc.send({ type: "UPLOAD", data: new Uint8Array([1, 2, 3, 4]) });

UPLOAD(msg) {
  msg.data;            // Uint8Array(4), the array itself
  [...msg.data];       // [1, 2, 3, 4]
  msg.data.byteLength; // 4
}

The far side receives the same kind of array that was sent, with its octets in it. A typed array is the one eager value: the reference states the length and the element kind, the sender is asked for the octets as soon as it arrives, and the message itself is held until the last one is in. A handler that never awaits reads the whole array, because the message is not delivered before it is whole.

The kind travels with the reference, so an Int32Array comes back an Int32Array and Float64Array a Float64Array. DataView and ArrayBuffer have no element kind to carry, so they arrive as Uint8Array; read those through a view of the type you meant.

The wait is spent by the connection, not by your code: while the octets of one message arrive, the messages behind it on that connection wait their turn. A transfer that fails — over the receiver's bound, or short of the length it stated — drops that message and reports why, rather than handing over a partly filled array.

Files

A Blob or File — from <input type="file">, fetch, new File(...), or the globals Node 20 and later provide — crosses as a file. The far side gets a RemoteBlob with the same facts:

proc.send({ type: "UPLOAD", shot: file });   // a File from an input

async UPLOAD(msg) {
  msg.shot.size;   // 81234
  msg.shot.type;   // "image/png"
  msg.shot.name;   // "shot.png"

  const all = await msg.shot.arrayBuffer();        // read the whole thing
  for await (const chunk of msg.shot.stream()) {   // or take it as it comes
    digest.update(chunk);
  }
}

Nothing is read until you ask, so a file nobody opens costs nothing on the wire. Read it once: the bytes are on the wire, not in the value, and a second read gets nothing. name is undefined when the sender's value had none, which is the case for a plain Blob.

A file on disk

An open file handle is not one of these values, so you describe it. A value counts as a file when it has size, arrayBuffer() and one of slice() or stream(). The connection reads the octets through slice() when it exists and through stream() otherwise, which is where a handle fits:

import { open } from "node:fs/promises";

const handle = await open("/data/big.iso");
const { size } = await handle.stat();

await proc.send({
  type: "UPLOAD",
  file: {
    size,
    name: "big.iso",
    arrayBuffer: async () => {
      const all = Buffer.alloc(size);
      await handle.read(all, 0, size, 0);
      return all.buffer.slice(all.byteOffset, all.byteOffset + all.length);
    },
    stream: () => handle.readableWebStream(),
  },
});

arrayBuffer() belongs to the shape because a file is a value you can read whole, and it is what a reader on this side would use; the connection itself takes the slice() or stream() path. A Buffer is a Uint8Array, so it travels as a typed array with no size or name of its own. Wrap it in a Blob when the far side needs to know what it is.

A file that arrived somewhere else

A RemoteBlob has the same shape, so it can be sent on without being read here:

async FORWARD(msg) {
  await this.emit({ type: "FORWARD", file: msg.file });
}

The second connection asks for the octets directly from the first. This holds while nobody has read the file yet; once it is read, there is nothing left to send.

Streams

An async iterable crosses as a stream:

async function* readInChunks() {
  for (const name of files) yield await readFile(name);
}

await proc.send({ type: "FOLLOW", feed: readInChunks() });

async FOLLOW(msg) {
  for await (const chunk of msg.feed) {
    digest.update(chunk);
  }
}

RemoteStream iterates as Uint8Array chunks. Where the runtime has ReadableStream, it also answers the platform's surface — getReader(), pipeTo(), tee(), asReadableStream() — over the same queue and the same one reader. bytesOf(stream) collects the whole thing into one array when that is what you need.

One consumer reads it. Leaving a for await loop cancels the stream: the sender's iterator gets return(), so a transfer you no longer want stops at the source instead of running to the end. Nothing flows before someone reads — the reader's demand grants credit, and the writer sends whole chunks against it.

A ReadableStream is itself an async iterable in browsers and in Node, so it can be sent as it is.

Pacing and limits

The defaults are 64 KiB per chunk, a 256 KiB credit window and 64 MiB per stream. A slow reader slows the writer; nothing is written ahead of demand, and each chunk carries a checksum the reader verifies as it arrives.

The bound belongs to the receiving side. A typed array or a file states its size on the reference, so it is refused before a chunk is asked for, and the message carrying it is dropped. A stream states no length, so it is failed once the octets pass the bound. A server sets its own limits with streams on the reactor:

const running = reactor(listener, app, {
  streams: {
    chunkBytes: 32 * 1024,
    windowBytes: 128 * 1024,
    maxBytes: 8 * 1024 * 1024,
  },
});

The octets travel beside the frames, so the transport has to carry them. The WebSocket and worker transports do. A connection built only of lines cannot, and a store over it reports canCarryStreams === false.

A program's own output

A payload that prints announces what it printed as two streams named stdout and stderr. The side holding the writing end announces them, the text spends no frames of its own, and the client reads them into onOutput:

const app = remoteClient("job", spawner, {
  onOutput: (name, text) => process.stdout.write(`[${name}] ${text}`),
});

What the payload printed before the other end could listen is replayed as the stream's first chunk. Any other name belongs to the application that holds the connection: StreamStore is exported, and a party that relays a connection it does not speak holds its own (new StreamStore(channel, parity, { attach: false })) and passes frames in through handle(). Such a store reads names with named() and onNamed().