Building your own transport

Remote actors are transport-agnostic: the seam (serveRemoteActor and remoteClient) speaks only in a Channel of decoded frame objects, plus the frame guards. Protocol, transport, and spawner are all pluggable, so the same actor can run over Kafka, a Unix socket, WebRTC, or an in-memory pipe once you write one transport and a spawner.

The three layers

The shipped FIFO, WebSocket, and worker transports are three concrete instances of this pattern, and a useful reference for writing your own.

Channel vs StringTransport

Frames are plain objects carrying one $-key each ($proto, $init, $state, $msg, $exit). Two interfaces move them:

// decoded frame objects — the seam's currency (also exported as `Channel`)
interface FrameChannel {
  send(frame: Record<string, unknown>): void | Promise<void>;
  onMessage(handler: (frame: Record<string, unknown>) => void): () => void;
  onClose(handler: () => void): void;
  close(): Promise<void>;
  // only on a connection that carries octets beside its lines
  sendSegment?(frame: Record<string, unknown>, bytes: Uint8Array): void | Promise<void>;
  onSegment?(handler: (frame: Record<string, unknown>, bytes: Uint8Array) => void): () => void;
}

// encoded frame strings — for byte streams (Kafka, TCP, sockets, fifos)
interface StringTransport {
  send(frame: string): void | Promise<void>;
  onMessage(handler: (frame: string) => number | void): void;
  onBytes(handler: (bytes: Uint8Array) => void): void;
  removeHandler(): void;
  sendBytes(bytes: Uint8Array): void | Promise<void>;
  onClose(handler: () => void): void;
  close(): Promise<void>;
}

onMessage answers how many octets follow the line, or nothing when none do. Those octets reach onBytes whole, however the medium split them, and the next line starts after them. A medium that only ever moves lines can answer nothing every time; the seam's sendSegment and onSegment then stay unavailable.

json1Channel(transport) lifts a StringTransport into a Channel with JSON encoding (json.v1). A medium with native message boundaries and structured clone (such as postMessage) can implement Channel directly and skip JSON — that's what WorkerTransport does (clone.v1).

A transport

For a byte stream, implement StringTransport: frame each outgoing message and split the incoming stream back into frames. Kafka or a Unix socket differs only in how the bytes move:

import type { StringTransport } from "posipaki/remote";

type ByteStream = {
  write(data: string | Uint8Array): void;
  on(event: string, cb: (chunk?: string | Uint8Array) => void): void;
  end(): void;
};

class NewlineStreamTransport implements StringTransport {
  private handler: ((line: string) => number | void) | null = null;
  private byteHandler: ((bytes: Uint8Array) => void) | null = null;
  private closeHandler: (() => void) | null = null;
  private buffer = Buffer.alloc(0);
  /** How many octets the last line said would follow it, or 0. */
  private want = 0;

  constructor(private stream: ByteStream) {
    stream.on("data", (chunk = "") => this.feed(Buffer.from(chunk)));
    stream.on("close", () => this.closeHandler?.());
  }

  private feed(chunk: Buffer) {
    this.buffer = Buffer.concat([this.buffer, chunk]);
    for (;;) {
      if (this.want > 0) {
        if (this.buffer.length < this.want) return;
        this.byteHandler?.(this.buffer.subarray(0, this.want));
        this.buffer = this.buffer.subarray(this.want);
        this.want = 0;
        continue;
      }
      const nl = this.buffer.indexOf(0x0a);
      if (nl === -1) return;
      const line = this.buffer.subarray(0, nl).toString("utf8");
      this.buffer = this.buffer.subarray(nl + 1);
      this.want = this.handler?.(line) ?? 0;
    }
  }

  send(frame: string) {
    this.stream.write(frame.endsWith("\n") ? frame : frame + "\n");
  }
  sendBytes(bytes: Uint8Array) {
    this.stream.write(bytes);
  }
  onMessage(handler: (line: string) => number | void) {
    this.handler = handler;
  }
  onBytes(handler: (bytes: Uint8Array) => void) {
    this.byteHandler = handler;
  }
  removeHandler() {
    this.handler = null;
  }
  onClose(handler: () => void) {
    this.closeHandler = handler;
  }
  async close() {
    this.stream.end();
  }
}

The shipped FifoUtf8NlineTransport is the full version: it also handles errors, makes close idempotent and enforces a single handler.

A spawner

A spawner obtains the transport and performs the $proto handshake, keeping the version string out of the seam. The server spawner sends it; the client spawner awaits and validates it:

import { json1Channel, VERSION, isProto } from "posipaki/remote";

// server: () => Promise<Channel>
const serverSpawner = async () => {
  const channel = json1Channel(new NewlineStreamTransport(acceptConnection()));
  await channel.send({ $proto: VERSION });
  return channel;
};

// client: (args) => Promise<Channel>
const clientSpawner = async (args) => {
  const channel = json1Channel(new NewlineStreamTransport(openConnection(args)));
  const frame = await new Promise((resolve) => {
    const off = channel.onMessage((f) => {
      off();
      resolve(f);
    });
  });
  if (!isProto(frame) || frame.$proto !== VERSION) {
    throw new Error("unsupported protocol");
  }
  return channel;
};

Wire it into the seam

import { serveRemoteActor, remoteClient } from "posipaki/remote";

// server process
serveRemoteActor(actor, serverSpawner);

// client process
const proxy = remoteClient("myactor", clientSpawner);
const proc = await proxy.spawn(args);
proc.send({ type: "PING", count: 1 });

remoteClient returns a normal ActorDefinition, so spawn, state, and wait behave like any other actor. serveRemoteActor takes the ActorDefinition (not a bare function), so plugins and reflection survive the trip.

Structured-clone transports

If your medium moves objects rather than bytes, implement Channel directly and handshake with WORKER_VERSION (clone.v1) instead of VERSION (json.v1).

Handshake reference