Class: IngestionPipeline

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:170

Turns ingested events into entity updates, in every scene bound to it, according to one or more EventMappings.

This is the whole data layer, with no dependency on an Agent, a Transport or a session: you bind scenes to it and push events in. That makes IngestionPipeline.ingest the primary verb of the SDK's data module — a mapping can be exercised from a test, a webhook, a REST handler or a replay tool without a broker anywhere in sight:

const pipeline = new IngestionPipeline({ mappings });
pipeline.bind({ scene });
await pipeline.ingest({ channel: "uagv/v2/m/AGV-1/visualization", payload });

For each event: pick the matching mappings; per mapping, (opt.) validate the payload against its schema, call its updates function to get one entity update or several — then, in each bound scene, resolve every id to an entity (found or spawned per the mapping's entities) and apply its update, skipping redundant component writes.

To drive the scenes an Agent is attached to, and to feed the pipeline from a transport, use SceneIngestion — it owns a pipeline and binds/unbinds sessions for you.

Constructors

Constructor

new IngestionPipeline(__namedParameters: IngestionPipelineOptions): IngestionPipeline;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:217

Parameters

Parameter	Type
__namedParameters	IngestionPipelineOptions

Returns

IngestionPipeline

Throws

If no mapping is given, or a mapping is malformed — its entities declaring none of the four strategies, or its updates not being a function. Both are unreachable from TypeScript; they are what a JavaScript consumer gets instead of a compile error.

Accessors

stats

Get Signature

get stats(): IngestionStats | null;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:247

What the pipeline has done so far — events in, updates out, and why anything was dropped. null when the pipeline was built with stats: false.

Returns

IngestionStats | null

---

continuationCount

Get Signature

get continuationCount(): number;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:264

How many ContinuousUpdates are installed across every mapping — the answer to "the stream is fine, so why is nothing moving?".

Also readable as stats.continuations_active; this getter works with stats: false too, and is what SceneIngestion checks to know whether its clock has anything to advance.

Returns

number

---

boundSceneCount

Get Signature

get boundSceneCount(): number;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:283

The number of scenes currently bound.

Returns

number

Methods

bind()

bind(__namedParameters: {
  scene: Scene;
}): PipelineBinding;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:294

Attach a scene: ingested events start driving its entities. Each scene gets its own resolvers, because a resolution (and its cache) is only meaningful against one scene.

Binding the same scene twice returns two independent bindings; unbind the one you no longer want.

Parameters

Parameter	Type
__namedParameters	{ scene: Scene; }
__namedParameters.scene	Scene

Returns

PipelineBinding

---

clearContinuations()

clearContinuations(__namedParameters?: {
  id?: string;
}): void;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:332

Stop the ContinuousUpdates currently installed — one id's, or every one of them. The entities keep their last value, and a later event carrying the same id installs a fresh motion.

The answer to "the broker died, why is the scene still moving?". Nothing expires a motion on its own, because "no event replaced it" is not the same fact as "the stream is dead" — one topic may carry payloads of several shapes. SceneIngestion calls this when it stops its sources.

The per-entity state is not dropped, so a motion started again picks up where this one stopped rather than snapping back.

Parameters

Parameter	Type
__namedParameters	{ id?: string; }
__namedParameters.id?	string

Returns

void

---

unbindAll()

unbindAll(): void;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:345

Detach every bound scene.

Returns

void

---

ingest()

ingest(event: IngestEvent): Promise<void>;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:357

Push one event through the pipeline: every mapping whose selectors match it drives the entities it addresses, in every bound scene.

Never throws — a mapping that throws is reported to onError and the stream continues.

Parameters

Parameter	Type
event	IngestEvent

Returns

Promise<void>

---

tick()

tick(elapsed_seconds: number): Promise<void>;

Defined in: livelink.clients/livelink.agent/sources/data/IngestionPipeline.ts:407

Advance every ContinuousUpdate currently installed, and write what they produce.

This is what keeps a machine moving between two messages: an event that reported a rate installed a continuation, and each tick asks it where the entity is now. It owns no timer — the caller decides the cadence, which is what makes a moving scene reproducible from a test:

await pipeline.ingest(event); // "turning at 90 rpm"
await pipeline.tick(0.5); // half a second later, wherever that puts it

SceneIngestion calls this on its own interval, so a consumer using it has nothing to do. A tick is deliberately not an event: it leaves events_received and last_event_at alone, so those keep answering "is data still arriving?" while the scene moves.

Never throws — a continuation that throws is reported to onError and dropped.

Parameters

Parameter	Type
elapsed_seconds	number

Returns

Promise<void>