Write Your Own Transport

The bundled transports cover four protocols. Everything else — Kafka, AMQP, Modbus, a serial port, a WebSocket feed, a REST webhook, a database poll — plugs into exactly the same pipeline through two interfaces of one method each.

The seam

// Implement these interfaces to create a custom transport.
// API reference: /references/livelink.agent/interfaces/Transport

interface Transport {
    start(): Promise<void>; // connect and begin forwarding; reject if it cannot be established
    stop(): Promise<void>; // disconnect; must be safe when never started
}

interface EventSink {
    ingest(event: IngestEvent): void | Promise<void>; // must not throw
}

There is no base class to extend and nothing to register at import time. A transport is any object with start and stop; by convention it takes its config and the sink it pushes into as constructor arguments.

Everything downstream — mappings, entity resolution, write deduplication, statistics — is already written and does not care where events came from.

A worked example

A WebSocket feed, which is the shape most custom sources end up having:

WebSocketTransport.ts

import type { EventSink, IngestEvent, Transport } from "@3dverse/livelink-agent";

export type WebSocketTransportConfig = {
    url: string;
    /** Channel given to events whose payload carries no topic of its own. */
    default_channel?: string;
};

export class WebSocketTransport implements Transport {
    readonly #config: WebSocketTransportConfig;
    readonly #sink: EventSink;
    #socket: WebSocket | null = null;

    constructor(config: WebSocketTransportConfig, sink: EventSink) {
        this.#config = config;
        this.#sink = sink;
    }

    async start(): Promise<void> {
        // Resolve once the source is live, reject if it cannot be established:
        // the ingestion awaits this, reports a rejection on `on-error`, and
        // retries the source when the next session binds.
        await new Promise<void>((resolve, reject) => {
            const socket = new WebSocket(this.#config.url);
            socket.addEventListener("open", () => resolve());
            socket.addEventListener("error", () => reject(new Error(`Cannot reach ${this.#config.url}`)));
            socket.addEventListener("message", event => void this.#handleMessage(event.data));
            this.#socket = socket;
        });
    }

    async stop(): Promise<void> {
        // Safe when never started.
        this.#socket?.close();
        this.#socket = null;
    }

    async #handleMessage(data: string): Promise<void> {
        let message: { topic?: string; ts?: number; body: unknown };
        try {
            message = JSON.parse(data);
        } catch {
            console.warn("[ws-transport] Invalid JSON");
            return;
        }

        const event: IngestEvent = {
            channel: message.topic ?? this.#config.default_channel ?? "websocket",
            payload: message.body,
            received_at: new Date(),
            ...(message.ts ? { source_timestamp: new Date(message.ts) } : {}),
            metadata: { transport: "websocket" },
        };

        try {
            // A failure to handle one event must never take the transport down.
            await this.#sink.ingest(event);
        } catch (error) {
            console.error("[ws-transport] Sink failed:", error);
        }
    }
}

Filling in the event

A transport's real job is decoding, and the envelope is where that work lands:

Field	What to put in it
channel	The source's routing key, if it has one. This is what mappings select on
payload	The decoded message. JSON-decoded, or whatever structure your protocol yields
received_at	new Date() — when this process saw it
source_timestamp	The source's own clock, when it reports one. Prefer it for staleness checks
metadata	Envelope fields with nowhere else to go — offsets, QoS, sequence numbers, partition ids

If your source has no routing key, use a constant channel and let mappings select on the payload with when, the way Azure Event Hubs does.

Never interpret the payload in a transport. Deciding what a value means is the mapping's job — that separation is why the same transport serves every use case.

Three ways to plug it in

1. As a factory (start here)

Pass a function receiving the sink. The ingestion still owns the lifecycle: started lazily on the first bound session, stopped with the ingestion.

const ingestion = new SceneIngestion({
    agent,
    pipeline,
    sources: [sink => new WebSocketTransport({ url: "wss://feed.example.com" }, sink)],
});

This is the right default for a one-off transport in one application.

2. As a registered kind

Register it on the process-wide registry and it becomes nameable by string, exactly like the bundled kinds:

import { defaultTransportRegistry } from "@3dverse/livelink-agent";

defaultTransportRegistry.register({
    kind: "websocket",
    factory: (config, sink) => new WebSocketTransport(config as WebSocketTransportConfig, sink),
});

sources: [{ kind: "websocket", config: { url: "wss://feed.example.com" } }],

Do this when several ingestions need the transport, or — the real reason — when the source has to be configurable without a rebuild. A registered kind means a { kind, config } pair survives a round trip through a config file, a database row, or an HTTP request body.

Registering a kind that already exists throws, so kinds stay unambiguous.

3. Own it yourself

SceneIngestion is an EventSink. Anything that can call a method can drive the scene:

await ingestion.start();

myFeed.on("data", async message => {
    await ingestion.ingest({ channel: message.topic, payload: message.body });
});

Reach for this when:

• The source must run before any viewer connects. Configured sources start lazily, on the first bound session. A transport you own starts when you start it.
• The events do not come from a stream at all — a webhook, a REST handler, a scheduled poll, a UI control replaying a recording one frame at a time.
• The source has a lifecycle of its own you would rather not hand over.

An Express webhook driving a scene

app.post("/telemetry", async (request, response) => {
    await ingestion.ingest({
        channel: `devices/${request.body.deviceId}/telemetry`,
        payload: request.body,
    });
    response.sendStatus(202);
});

The pipeline underneath works the same way, with no agent at all — pipeline.ingest(event) against any bound scene.

The contract, in full

Four rules, all of them about failure:

1. start() resolves once the source is live, and rejects if it cannot be established. Sources are started sequentially and awaited, so do not retry forever inside start — a rejection is reported on the ingestion's on-error and retried when the next session binds, which is the loop that survives a source being down at boot.
2. stop() must be safe when start() was never called, and when it failed. It is called on every shutdown path.
3. ingest must never throw at the transport. Wrap the call: one bad event must not take the connection down.
4. Decode, do not interpret. Ship the message and its envelope; let the mapping decide what it means.

Wrapping the sink

Because the sink is one method, anything that observes or transforms the stream is a few lines and needs no SDK support. Tap it for a live trace:

function observedSink({ sink, onEvent }: { sink: EventSink; onEvent: (event: IngestEvent) => void }): EventSink {
    return {
        ingest(event) {
            onEvent(event);
            return sink.ingest(event);
        },
    };
}

sources: [sink => new WebSocketTransport(config, observedSink({ sink, onEvent: trace }))],

The same shape covers recording a stream for playback, throttling a source that outruns the scene, and fanning one transport out to several pipelines.