Example 34: Intake Stream Source
What It Is
Intake Stream Source is the Cartographer’s open-input pattern: each event type enters as a DAG entrypoint label, every label targets intake-gather, and the gather writes one merged async stream into state.sources for process-stream.
The graph shows the intake explicitly. There is no seed pre-phase and no scatter before the gather.
How It Works
The scheduler seeds one gather record per entrypoint label because all five labels target the same GatherNode. The source-intake gather opens the matching event-type stream from state.eventConfig, round-robins the streams into a single AsyncIterable<SourcePayload>, and assigns that stream to state.sources. The scatter reads from that field using the normal source accessor and pulls the next item only when concurrency and reservoir capacity allow it.
This keeps streaming as a DAG concern without a special streaming placement. The intake is a gather, the processor is a scatter, and both are visible in JSON-LD.
Diagrams, Examples, and Outputs
DAG registration and diagram
The Cartographer enters through five data-type entrypoints. Each entrypoint targets intake-gather; only after the gather completes does process-stream consume state.sources.
Cartographer streaming source DAG
5 placements{
"@context": {
"@version": 1.1,
"name": {
"@id": "https://noocodec.dev/ontology/dag/name"
},
"version": {
"@id": "https://noocodec.dev/ontology/dag/version"
},
"entrypoints": {
"@id": "https://noocodec.dev/ontology/dag/entrypoints",
"@container": "@index"
},
"nodes": {
"@id": "https://noocodec.dev/ontology/dag/nodes",
"@container": "@set"
},
"outputs": {
"@id": "https://noocodec.dev/ontology/dag/outputs"
},
"node": {
"@id": "https://noocodec.dev/ontology/dag/node"
},
"dag": {
"@id": "https://noocodec.dev/ontology/dag/dag"
},
"body": {
"@id": "https://noocodec.dev/ontology/dag/body"
},
"source": {
"@id": "https://noocodec.dev/ontology/dag/source"
},
"sources": {
"@id": "https://noocodec.dev/ontology/dag/sources",
"@container": "@index"
},
"itemKey": {
"@id": "https://noocodec.dev/ontology/dag/itemKey"
},
"execution": {
"@id": "https://noocodec.dev/ontology/dag/execution"
},
"concurrency": {
"@id": "https://noocodec.dev/ontology/dag/concurrency"
},
"throttle": {
"@id": "https://noocodec.dev/ontology/dag/throttle"
},
"reservoir": {
"@id": "https://noocodec.dev/ontology/dag/reservoir"
},
"gather": {
"@id": "https://noocodec.dev/ontology/dag/gather"
},
"dagReference": {
"@id": "https://noocodec.dev/ontology/dag/dagReference",
"@type": "@id"
},
"DagReference": {
"@id": "https://noocodec.dev/ontology/dag/DagReference"
},
"from": {
"@id": "https://noocodec.dev/ontology/dag/from"
},
"path": {
"@id": "https://noocodec.dev/ontology/dag/path"
},
"candidates": {
"@id": "https://noocodec.dev/ontology/dag/candidates",
"@container": "@set"
},
"candidateDag": {
"@id": "https://noocodec.dev/ontology/dag/candidateDag",
"@type": "@id"
},
"selectedDag": {
"@id": "https://noocodec.dev/ontology/dag/selectedDag",
"@type": "@id"
},
"resultField": {
"@id": "https://noocodec.dev/ontology/dag/resultField"
},
"policy": {
"@id": "https://noocodec.dev/ontology/dag/policy"
},
"reducer": {
"@id": "https://noocodec.dev/ontology/dag/reducer"
},
"outcome": {
"@id": "https://noocodec.dev/ontology/dag/outcome"
},
"phase": {
"@id": "https://noocodec.dev/ontology/dag/phase"
},
"stateMapping": {
"@id": "https://noocodec.dev/ontology/dag/stateMapping"
},
"container": {
"@id": "https://noocodec.dev/ontology/dag/container"
},
"DAG": {
"@id": "https://noocodec.dev/ontology/dag/DAG"
},
"Placement": {
"@id": "https://noocodec.dev/ontology/dag/Placement"
},
"SingleNode": {
"@id": "https://noocodec.dev/ontology/dag/SingleNode"
},
"ScatterNode": {
"@id": "https://noocodec.dev/ontology/dag/ScatterNode"
},
"EmbeddedDAGNode": {
"@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
},
"GatherNode": {
"@id": "https://noocodec.dev/ontology/dag/GatherNode"
},
"TerminalNode": {
"@id": "https://noocodec.dev/ontology/dag/TerminalNode"
},
"PhaseNode": {
"@id": "https://noocodec.dev/ontology/dag/PhaseNode"
}
},
"@id": "urn:noocodec:dag:cartographer",
"@type": "DAG",
"name": "dag:cartographer",
"version": "1.0",
"entrypoints": {
"position-ping": "urn:noocodec:dag:cartographer/node/intake-gather",
"facility-scan": "urn:noocodec:dag:cartographer/node/intake-gather",
"sensor-reading": "urn:noocodec:dag:cartographer/node/intake-gather",
"customs-event": "urn:noocodec:dag:cartographer/node/intake-gather",
"delivery-confirmation": "urn:noocodec:dag:cartographer/node/intake-gather"
},
"nodes": [
{
"@id": "urn:noocodec:dag:cartographer/node/intake-gather",
"@type": "GatherNode",
"name": "dag:cartographer/node/intake-gather",
"sources": {
"urn:noocodec:dag:cartographer/entrypoint/position-ping": {},
"urn:noocodec:dag:cartographer/entrypoint/facility-scan": {},
"urn:noocodec:dag:cartographer/entrypoint/sensor-reading": {},
"urn:noocodec:dag:cartographer/entrypoint/customs-event": {},
"urn:noocodec:dag:cartographer/entrypoint/delivery-confirmation": {}
},
"gather": {
"strategy": "source-intake"
},
"outputs": {
"success": "urn:noocodec:dag:cartographer/node/process-stream",
"error": "urn:noocodec:dag:cartographer/node/process-stream",
"empty": "urn:noocodec:dag:cartographer/node/done"
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/process-stream",
"@type": "ScatterNode",
"name": "dag:cartographer/node/process-stream",
"source": "sources",
"body": {
"dag": "urn:noocodec:dag:stream-event"
},
"outputs": {
"all-success": "urn:noocodec:dag:cartographer/node/fold-insights",
"partial": "urn:noocodec:dag:cartographer/node/fold-insights",
"all-error": "urn:noocodec:dag:cartographer/node/fold-insights",
"empty": "urn:noocodec:dag:cartographer/node/summarize"
},
"itemKey": "source-payload",
"reducer": "aggregate",
"execution": {
"mode": "reservoir",
"concurrency": 16,
"reservoir": {
"keyField": "eventType",
"capacity": 1000
}
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/fold-insights",
"@type": "GatherNode",
"name": "dag:cartographer/node/fold-insights",
"sources": {
"urn:noocodec:dag:cartographer/node/process-stream": {}
},
"gather": {
"strategy": "insights-fold"
},
"outputs": {
"success": "urn:noocodec:dag:cartographer/node/summarize",
"error": "urn:noocodec:dag:cartographer/node/summarize",
"empty": "urn:noocodec:dag:cartographer/node/summarize"
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/summarize",
"@type": "SingleNode",
"name": "dag:cartographer/node/summarize",
"node": "urn:noocodec:node:summarize",
"outputs": {
"success": "urn:noocodec:dag:cartographer/node/done"
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/done",
"@type": "TerminalNode",
"name": "dag:cartographer/node/done",
"outcome": "completed"
}
]
}Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
%% dag:cartographer (v1.0)
entry_position-ping(["position-ping"])
entry_position-ping --> urn_noocodec_dag_cartographer/node/intake-gather
entry_facility-scan(["facility-scan"])
entry_facility-scan --> urn_noocodec_dag_cartographer/node/intake-gather
entry_sensor-reading(["sensor-reading"])
entry_sensor-reading --> urn_noocodec_dag_cartographer/node/intake-gather
entry_customs-event(["customs-event"])
entry_customs-event --> urn_noocodec_dag_cartographer/node/intake-gather
entry_delivery-confirmation(["delivery-confirmation"])
entry_delivery-confirmation --> urn_noocodec_dag_cartographer/node/intake-gather
urn_noocodec_dag_cartographer/node/intake-gather{"dag_cartographer/node/intake-gather"}
urn_noocodec_dag_cartographer/node/intake-gather -->|success| urn_noocodec_dag_cartographer/node/process-stream
urn_noocodec_dag_cartographer/node/intake-gather -->|error| urn_noocodec_dag_cartographer/node/process-stream
urn_noocodec_dag_cartographer/node/intake-gather -->|empty| urn_noocodec_dag_cartographer/node/done
urn_noocodec_dag_cartographer/node/process-stream[/"dag:cartographer/node/process-stream ▣ eventType ×1000"/]
urn_noocodec_dag_cartographer/node/process-stream -->|all-success| urn_noocodec_dag_cartographer/node/fold-insights
urn_noocodec_dag_cartographer/node/process-stream -->|partial| urn_noocodec_dag_cartographer/node/fold-insights
urn_noocodec_dag_cartographer/node/process-stream -->|all-error| urn_noocodec_dag_cartographer/node/fold-insights
urn_noocodec_dag_cartographer/node/process-stream -->|empty| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/fold-insights{"dag_cartographer/node/fold-insights"}
urn_noocodec_dag_cartographer/node/fold-insights -->|success| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/fold-insights -->|error| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/fold-insights -->|empty| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/summarize["dag:cartographer/node/summarize"]
urn_noocodec_dag_cartographer/node/summarize -->|success| urn_noocodec_dag_cartographer/node/done
urn_noocodec_dag_cartographer/node/done((("dag:cartographer/node/done")))
classDef reservoir fill:#1e3a5f,stroke:#3b82f6,color:#bfdbfe
class urn_noocodec_dag_cartographer/node/process-stream reservoirThe stream source is not a hidden host-side setup step. JSON-LD shows the graph shape: data-type entrypoint labels, first-class gather, then the worker-capable scatter.
Run
bash
npm run docs:devOpen The Cartographer and run the stream.
What It Lets You Do
Intake stream sources let entrypoints supply async data without materializing the full input collection first. Use this when different domains or data types should enter a DAG independently before converging into one processing stream.
For applications, the useful property is clarity: intake shape is graph-visible, while the scatter still controls how quickly work is drained through concurrency and reservoir settings.
Code Samples
CartographerSourceIntake owns per-type stream construction:
ts
/**
* sourceIntake: stream construction helpers for Cartographer's open intake gather.
*
* The intake gather is entrypoint-driven: each data-type entrypoint is a
* canonical entrypoint IRI, and the gather/source helpers open the matching
* typed source stream directly.
*/
import type { CartographerState } from '../CartographerState.ts';
import type { SourcePayload } from '../entities/SourcePayload.ts';
import type { CanonicalEventVariant } from '../entities/index.ts';
import { EventStreamSource } from '../services/EventStreamSource.ts';
import { CARTOGRAPHER_IRIS } from '../cartographerIds.ts';
import type { GatherRecordType } from '@studnicky/dagonizer/contracts';
import type { NodeStateInterface } from '@studnicky/dagonizer/types';
type CartographerEventType = CanonicalEventVariant['eventType'];
type CartographerIntakeState = NodeStateInterface & Pick<CartographerState, 'eventConfig' | 'streamCount'>;
class SourcePayloadStream {
private constructor() { /* static-only */ }
static async *roundRobin(streams: readonly AsyncIterable<SourcePayload>[]): AsyncIterable<SourcePayload> {
const active = streams.map((stream) => stream[Symbol.asyncIterator]());
while (active.length > 0) {
for (let i = 0; i < active.length;) {
const step = await active[i]?.next();
if (step === undefined || step.done === true) {
active.splice(i, 1);
continue;
}
yield step.value;
i++;
}
}
}
static async *skip(stream: AsyncIterable<SourcePayload>, count: number): AsyncIterable<SourcePayload> {
let skipped = 0;
for await (const item of stream) {
if (skipped < count) {
skipped++;
continue;
}
yield item;
}
}
static async *empty(): AsyncIterable<SourcePayload> {
return;
}
}
export class CartographerSourceIntake {
private constructor() { /* static-only */ }
static isState(state: NodeStateInterface): state is CartographerIntakeState {
const eventConfig = Reflect.get(state, 'eventConfig');
return eventConfig !== null
&& typeof eventConfig === 'object'
&& typeof Reflect.get(state, 'streamCount') === 'number';
}
static streamFor(state: CartographerIntakeState, eventType: CartographerEventType): AsyncIterable<SourcePayload> {
const totalCount = state.streamCount > 0 ? state.streamCount : undefined;
return CartographerSourceIntake.filterType(
EventStreamSource.streamTyped(state.eventConfig, totalCount),
eventType,
);
}
static mergedFor(state: CartographerState, resumeAfter: number = 0): AsyncIterable<SourcePayload> {
const streams = CARTOGRAPHER_IRIS.intakeEventTypes.map((eventType) =>
CartographerSourceIntake.streamFor(state, eventType),
);
const merged = SourcePayloadStream.roundRobin(streams);
return resumeAfter > 0 ? SourcePayloadStream.skip(merged, resumeAfter) : merged;
}
static mergeRecords(
records: readonly GatherRecordType[],
state: NodeStateInterface,
): AsyncIterable<SourcePayload> {
if (!CartographerSourceIntake.isState(state)) return SourcePayloadStream.empty();
const streams: AsyncIterable<SourcePayload>[] = [];
for (const source of CARTOGRAPHER_IRIS.intakeEventTypes) {
const record = records.find((candidate) => CartographerSourceIntake.sourceType(candidate.source) === source);
if (record === undefined) continue;
streams.push(CartographerSourceIntake.streamFor(state, source));
}
return SourcePayloadStream.roundRobin(streams);
}
private static sourceType(source: string): CartographerEventType | null {
const marker = '/entrypoint/';
const index = source.indexOf(marker);
if (index < 0) return null;
const label = decodeURIComponent(source.slice(index + marker.length));
return CARTOGRAPHER_IRIS.intakeEventTypes.includes(label as CartographerEventType)
? (label as CartographerEventType)
: null;
}
private static async *filterType(
stream: AsyncIterable<SourcePayload>,
eventType: CartographerEventType,
): AsyncIterable<SourcePayload> {
for await (const item of stream) {
if (item.eventType === eventType) yield item;
}
}
}SourceIntakeGather merges the entrypoint records into the stream that process-stream consumes:
ts
export class SourceIntakeGather extends GatherStrategy {
readonly name = 'source-intake';
readonly '@id' = 'urn:noocodec:node:source-intake';
override initial(
_config: GatherConfigType,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
accessor.set(state, 'sources', []);
}
override reduce(
_config: GatherConfigType,
batch: Parameters<GatherStrategy['reduce']>[1],
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
const records: GatherRecordType[] = [];
for (const item of batch) records.push(item.state);
accessor.set(state, 'sources', CartographerSourceIntake.mergeRecords(records, state));
}
}
GatherStrategies.register(new SourceIntakeGather());Details for Nerds
- Bounded streaming source. The scatter consumes an
AsyncIterable<SourcePayload>without materializing the whole feed. - Back-pressure by pull rate. The scatter controls drain speed through concurrency and reservoir settings.
- Graph-visible intake. Switching from one producer to many producers changes DAG topology intentionally: the gather is the intake point.
- Runnable browser ownership. The Cartographer page exposes event count, worker pool size, and batch capacity controls for the same stream.
Related Concepts
- Example 17: Async source - async-iterable scatter source in the same Cartographer DAG
- Example 35: Stream resume - durable cursor resume on the Cartographer stream
- Streaming Producers - full streaming API: driven, fanIn, resumable, DagStreamProducer