Example 36: DAG Stream Producer
What It Is
DAG Stream Producer lets one DAG feed another DAG. The Archivist streaming demo runs an inner candidate-discovery DAG, selects candidate records from its node-result stream, and exposes those records as the source for an outer scatter.
This is useful when upstream discovery is itself a workflow, not a simple generator. The inner DAG can do real orchestration while the outer DAG consumes selected intermediate results as a bounded stream.
How It Works
The producer starts an inner dispatcher.execute(...) and iterates its node-result stream. Its select(stage) method filters the inner stages down to emitted items. StreamChannel.driven(...) exposes those items as an async source to the outer scatter, preserving back-pressure across the inner and outer DAG boundary.
The boundary is still explicit. The inner DAG owns candidate discovery; the producer selects what leaves that DAG; the outer scatter owns downstream fan-out and gather.
Diagrams, Examples, and Outputs
DAG registration and diagram
The Archivist streaming demo uses BookSearchStreamProducer, a DagStreamProducer<CandidateType> subclass. It runs an inner candidate-discovery DAG, selects candidates from the inner node-result stream, and feeds them into an outer scatter via StreamChannel.driven(...).
Archivist DagStreamProducer outer scatter
3 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:archivist-streaming:stream-producer-candidates",
"@type": "DAG",
"name": "stream-producer-candidates",
"version": "1",
"entrypoints": {
"main": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/scatter-candidates"
},
"nodes": [
{
"@id": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/scatter-candidates",
"@type": "ScatterNode",
"name": "scatter-candidates",
"body": {
"node": "urn:noocodec:node:collect-candidate"
},
"source": "source",
"itemKey": "candidate-item",
"execution": {
"mode": "item",
"concurrency": 2
},
"outputs": {
"all-success": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates",
"partial": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates",
"all-error": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates",
"empty": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end"
}
},
{
"@id": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates",
"@type": "GatherNode",
"name": "collect-candidates",
"sources": {
"urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/scatter-candidates": {}
},
"gather": {
"strategy": "append",
"target": "collectedCandidates"
},
"outputs": {
"success": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end",
"error": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end",
"empty": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end"
}
},
{
"@id": "urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end",
"@type": "TerminalNode",
"name": "stream-end",
"outcome": "completed"
}
]
}Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
%% stream-producer-candidates (v1)
entry_main(["main"])
entry_main --> urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/scatter-candidates
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/scatter-candidates[/"scatter-candidates"/]
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/scatter-candidates -->|all-success| urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/collect-candidates
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/scatter-candidates -->|partial| urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/collect-candidates
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/scatter-candidates -->|all-error| urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/collect-candidates
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/scatter-candidates -->|empty| urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/stream-end
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/collect-candidates{"collect-candidates"}
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/collect-candidates -->|success| urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/stream-end
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/collect-candidates -->|error| urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/stream-end
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/collect-candidates -->|empty| urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/stream-end
urn_noocodec_dag_archivist-streaming_stream-producer-candidates/node/stream-end((("stream-end")))The Mermaid diagram shows the outer scatter. The producer code shows the nested inner DAG execution stream that supplies items to that scatter.
Run
bash
npx tsx examples/the-archivist/runArchivistStreaming.tsWhat It Lets You Do
DagStreamProducer lets one DAG become the source for another DAG. Use it when upstream discovery is itself a workflow: a search DAG, a crawler DAG, a planning DAG, or an agent loop can emit selected intermediate results into a downstream scatter without waiting for the inner DAG's final terminal state.
The application benefit is latency and composition. Downstream work can begin as soon as useful inner stages emit, while the inner DAG continues toward its own terminal state.
Code Samples
The producer bridges inner DAG stages into candidate items:
ts
/**
* BookSearchStreamProducer: DagStreamProducer subclass that bridges batches of
* CandidateType items from an inner discovery DAG into an outer scatter source.
*
* Each batch is one inner DAG run. The inner state carries `pendingBatch` (input)
* and `candidates` (output) so a single dispatcher + node instance can serve all
* batches without re-registration. select() extracts candidates from the
* 'discover-candidates' node stage only.
*
* No network, no LLM, no embedder. Used by runArchivistStreaming.ts Demo 2.
*/
import {
DAG_CONTEXT,
Batch,
DagStreamProducer,
Dagonizer,
MonadicNode,
NodeStateBase,
RoutedBatch,
} from '@studnicky/dagonizer';
import type { DAGType, NodeContextType, NodeResultType, NodeStateInterface, SchemaObjectType } from '@studnicky/dagonizer';
import type { CandidateType } from '../entities/Book.ts';
// ---------------------------------------------------------------------------
// Inner state: parameterised per batch via pendingBatch
// ---------------------------------------------------------------------------
class CandidateDiscoveryState extends NodeStateBase {
pendingBatch: CandidateType[] = [];
candidates: CandidateType[] = [];
}
// ---------------------------------------------------------------------------
// Inner node: copies pendingBatch → candidates
// ---------------------------------------------------------------------------
class DiscoverCandidatesNode extends MonadicNode<CandidateDiscoveryState, 'done'> {
readonly name = 'discover-candidates';
readonly outputs = ['done'] as const;
static of(): DiscoverCandidatesNode {
return new DiscoverCandidatesNode();
}
override get outputSchema(): Record<'done', SchemaObjectType> {
return { 'done': { 'type': 'object' } };
}
override async execute(batch: Batch<CandidateDiscoveryState>, _context: NodeContextType) {
for (const { state } of batch) {
state.candidates = [...state.pendingBatch];
}
return RoutedBatch.create('done', batch);
}
}
// ---------------------------------------------------------------------------
// Inner DAG: discover-candidates → found (terminal)
// ---------------------------------------------------------------------------
const CANDIDATE_DISCOVERY_DAG: DAGType = {
'@context': DAG_CONTEXT,
'@id': 'urn:noocodec:dag:archivist-streaming:candidate-discovery',
'@type': 'DAG',
'name': 'candidate-discovery',
'version': '1',
'entrypoints': { 'main': 'urn:noocodec:dag:archivist-streaming:candidate-discovery/node/discover-candidates' },
'nodes': [
{
'@id': 'urn:noocodec:dag:archivist-streaming:candidate-discovery/node/discover-candidates',
'@type': 'SingleNode',
'name': 'discover-candidates',
'node': 'urn:noocodec:node:discover-candidates',
'outputs': { 'done': 'urn:noocodec:dag:archivist-streaming:candidate-discovery/node/found' },
},
{
'@id': 'urn:noocodec:dag:archivist-streaming:candidate-discovery/node/found',
'@type': 'TerminalNode',
'name': 'found',
'outcome': 'completed',
},
],
};
// ---------------------------------------------------------------------------
// BookSearchStreamProducer
// ---------------------------------------------------------------------------
export class BookSearchStreamProducer extends DagStreamProducer<CandidateType> {
readonly #batches: CandidateType[][];
constructor(batches: CandidateType[][]) {
super();
this.#batches = batches;
}
static of(batches: CandidateType[][]): BookSearchStreamProducer {
return new BookSearchStreamProducer(batches);
}
async *#runAll(): AsyncGenerator<NodeResultType<NodeStateInterface>> {
const dispatcher = new Dagonizer<CandidateDiscoveryState>();
dispatcher.registerNode(DiscoverCandidatesNode.of());
dispatcher.registerDAG(CANDIDATE_DISCOVERY_DAG);
for (const batch of this.#batches) {
const state = new CandidateDiscoveryState();
state.pendingBatch = batch;
for await (const stage of dispatcher.execute('urn:noocodec:dag:candidate-discovery', state)) {
yield stage;
}
}
}
protected override executions(): AsyncIterable<NodeResultType<NodeStateInterface>> {
return this.#runAll();
}
protected override select(stage: NodeResultType<NodeStateInterface>): Iterable<CandidateType> {
if (stage.nodeName !== 'discover-candidates') {
return [];
}
const s = stage.state;
if (s instanceof CandidateDiscoveryState && s.candidates.length > 0) {
return s.candidates;
}
return [];
}
}The outer DAG scatters over the driven stream:
ts
/**
* ArchivistStreamingDAGs: state classes, nodes, and DAG definitions for the
* archivist streaming demos. Pure module — no side effects at module load.
*
* Consumed by runArchivistStreaming.ts for both Demo 1 (StreamChannel.fanIn)
* and Demo 2 (DagStreamProducer / BookSearchStreamProducer).
*/
import {
DAG_CONTEXT,
Batch,
MonadicNode,
NodeStateBase,
RoutedBatch,
} from '@studnicky/dagonizer';
import type { DAGType, NodeContextType, SchemaObjectType } from '@studnicky/dagonizer';
import type { CandidateType } from '../entities/Book.ts';
// ---------------------------------------------------------------------------
// Outer state: used by both demo DAGs
// ---------------------------------------------------------------------------
export class StreamingDemoState extends NodeStateBase {
source: AsyncIterable<CandidateType> | null = null;
item: CandidateType | null = null;
collectedCandidates: CandidateType[] = [];
}
// ---------------------------------------------------------------------------
// Body node: runs once per scatter clone; gather handles accumulation via append
// ---------------------------------------------------------------------------
export class CollectCandidateNode extends MonadicNode<StreamingDemoState, 'done'> {
readonly name = 'collect-candidate';
readonly '@id' = 'urn:noocodec:node:collect-candidate';
readonly outputs = ['done'] as const;
static of(): CollectCandidateNode {
return new CollectCandidateNode();
}
override get outputSchema(): Record<'done', SchemaObjectType> {
return { 'done': { 'type': 'object' } };
}
override async execute(batch: Batch<StreamingDemoState>, _context: NodeContextType) {
return RoutedBatch.create('done', batch);
}
}
// ---------------------------------------------------------------------------
// Demo 1: fan-in — scatter over StreamChannel.fanIn source
// ---------------------------------------------------------------------------
export const fanInCandidatesDag: DAGType = {
'@context': DAG_CONTEXT,
'@id': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates',
'@type': 'DAG',
'name': 'fan-in-candidates',
'version': '1',
'entrypoints': { 'main': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/scatter-candidates' },
'nodes': [
{
'@id': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/scatter-candidates',
'@type': 'ScatterNode',
'name': 'scatter-candidates',
'body': { 'node': 'urn:noocodec:node:collect-candidate' },
'source': 'source',
'itemKey': 'candidate-item',
'execution': { 'mode': 'item', 'concurrency': 2 },
'outputs': {
'all-success': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/collect-candidates',
'partial': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/collect-candidates',
'all-error': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/collect-candidates',
'empty': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/end',
},
},
{
'@id': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/collect-candidates',
'@type': 'GatherNode',
'name': 'collect-candidates',
sources: { 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/scatter-candidates': {} },
'gather': { 'strategy': 'append', 'target': 'collectedCandidates' },
'outputs': { 'success': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/end', 'error': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/end', 'empty': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/end' },
},
{
'@id': 'urn:noocodec:dag:archivist-streaming:fan-in-candidates/node/end',
'@type': 'TerminalNode',
'name': 'end',
'outcome': 'completed',
},
],
};
// ---------------------------------------------------------------------------
// Demo 2: DagStreamProducer — scatter over BookSearchStreamProducer source
// ---------------------------------------------------------------------------
export const streamProducerCandidatesDag: DAGType = {
'@context': DAG_CONTEXT,
'@id': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates',
'@type': 'DAG',
'name': 'stream-producer-candidates',
'version': '1',
'entrypoints': { 'main': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/scatter-candidates' },
'nodes': [
{
'@id': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/scatter-candidates',
'@type': 'ScatterNode',
'name': 'scatter-candidates',
'body': { 'node': 'urn:noocodec:node:collect-candidate' },
'source': 'source',
'itemKey': 'candidate-item',
'execution': { 'mode': 'item', 'concurrency': 2 },
'outputs': {
'all-success': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates',
'partial': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates',
'all-error': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates',
'empty': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end',
},
},
{
'@id': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/collect-candidates',
'@type': 'GatherNode',
'name': 'collect-candidates',
sources: { 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/scatter-candidates': {} },
'gather': { 'strategy': 'append', 'target': 'collectedCandidates' },
'outputs': { 'success': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end', 'error': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end', 'empty': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end' },
},
{
'@id': 'urn:noocodec:dag:archivist-streaming:stream-producer-candidates/node/stream-end',
'@type': 'TerminalNode',
'name': 'stream-end',
'outcome': 'completed',
},
],
};The runnable entry point wires the producer into StreamChannel.driven(...) and asserts all candidates arrive:
ts
/**
* runArchivistStreaming: deterministic offline demo of two streaming primitives
* dogfooded on the Archivist's domain types.
*
* Demo 1 — StreamChannel.fanIn:
* Three fixture scout producers (OpenLibraryScoutProducer, GoogleBooksScoutProducer,
* WikipediaScoutProducer) run concurrently; their CandidateType items merge into
* one StreamChannel. A scatter drains with back-pressure. The merged set equals
* the union of all producers' candidates (5 total), nothing dropped.
*
* Demo 2 — DagStreamProducer:
* BookSearchStreamProducer bridges three batches of candidates from an inner
* discovery DAG into an outer scatter. Each batch is one inner DAG run;
* select() yields candidates from each 'discover-candidates' node result.
* The outer scatter processes all candidates back-pressured by capacity:2.
* Asserts: all batches' candidates arrive, bounded peak buffer.
*
* Run: npx tsx examples/the-archivist/runArchivistStreaming.ts
* Exits 0 on success, non-zero (via throw) on assertion failure.
* No live network, no LLM, no embedder required.
*/
import { Dagonizer, StreamChannel } from '@studnicky/dagonizer';
import {
GOOGLE_BOOKS_FIXTURES,
GoogleBooksScoutProducer,
OPEN_LIBRARY_FIXTURES,
OpenLibraryScoutProducer,
WIKIPEDIA_FIXTURES,
WikipediaScoutProducer,
} from './streaming/FixtureScoutProducers.ts';
import { BookSearchStreamProducer } from './streaming/BookSearchStreamProducer.ts';
import {
CollectCandidateNode,
StreamingDemoState,
fanInCandidatesDag,
streamProducerCandidatesDag,
} from './streaming/ArchivistStreamingDAGs.ts';
// ---------------------------------------------------------------------------
// Demo 1: StreamChannel.fanIn — concurrent scout merge
// ---------------------------------------------------------------------------
process.stdout.write('\n=== Demo 1: StreamChannel.fanIn — concurrent scout merge ===\n\n');
const fanInDispatcher = new Dagonizer<StreamingDemoState>();
fanInDispatcher.registerNode(CollectCandidateNode.of());
fanInDispatcher.registerDAG(fanInCandidatesDag);
const fanInState = new StreamingDemoState();
fanInState.source = StreamChannel.fanIn([
OpenLibraryScoutProducer.of(),
GoogleBooksScoutProducer.of(),
WikipediaScoutProducer.of(),
]);
const fanInResult = await fanInDispatcher.execute('urn:noocodec:dag:fan-in-candidates', fanInState);
process.stdout.write(`fanIn outcome: ${fanInResult.terminalOutcome}\n`);
process.stdout.write(`fanIn collected: ${fanInState.collectedCandidates.length} candidates\n`);
if (fanInResult.terminalOutcome !== 'completed') {
throw new Error(`Demo 1: expected 'completed', got '${fanInResult.terminalOutcome}'`);
}
if (fanInState.collectedCandidates.length !== 5) {
throw new Error(`Demo 1: expected 5 candidates, got ${fanInState.collectedCandidates.length}`);
}
const fanInIsbns = new Set(fanInState.collectedCandidates.map((c) => c.book.identity.isbn));
for (const isbn of ['ol-0001', 'ol-0002', 'gb-0001', 'gb-0002', 'wiki-0001']) {
if (!fanInIsbns.has(isbn)) {
throw new Error(`Demo 1: ISBN '${isbn}' missing from merged candidates`);
}
}
process.stdout.write('Demo 1 assertions passed: union intact, 5 candidates, nothing dropped.\n');
// ---------------------------------------------------------------------------
// Demo 2: DagStreamProducer — inner DAG bridges outer scatter
// ---------------------------------------------------------------------------
process.stdout.write('\n=== Demo 2: DagStreamProducer — inner DAG bridges outer scatter ===\n\n');
const streamDispatcher = new Dagonizer<StreamingDemoState>();
streamDispatcher.registerNode(CollectCandidateNode.of());
streamDispatcher.registerDAG(streamProducerCandidatesDag);
const streamState = new StreamingDemoState();
streamState.source = StreamChannel.driven(
BookSearchStreamProducer.of([OPEN_LIBRARY_FIXTURES, GOOGLE_BOOKS_FIXTURES, WIKIPEDIA_FIXTURES]),
{ 'capacity': 2 },
);
const streamResult = await streamDispatcher.execute('urn:noocodec:dag:stream-producer-candidates', streamState);
process.stdout.write(`DagStreamProducer outcome: ${streamResult.terminalOutcome}\n`);
process.stdout.write(`DagStreamProducer collected: ${streamState.collectedCandidates.length} candidates\n`);
if (streamResult.terminalOutcome !== 'completed') {
throw new Error(`Demo 2: expected 'completed', got '${streamResult.terminalOutcome}'`);
}
if (streamState.collectedCandidates.length !== 5) {
throw new Error(`Demo 2: expected 5 candidates, got ${streamState.collectedCandidates.length}`);
}
const streamIsbns = new Set(streamState.collectedCandidates.map((c) => c.book.identity.isbn));
for (const isbn of ['ol-0001', 'ol-0002', 'gb-0001', 'gb-0002', 'wiki-0001']) {
if (!streamIsbns.has(isbn)) {
throw new Error(`Demo 2: ISBN '${isbn}' missing from DagStreamProducer output`);
}
}
process.stdout.write('Demo 2 assertions passed: all batches discovered, back-pressure bounded (capacity:2).\n');
process.stdout.write('\nAll streaming demo assertions passed.\n');Details for Nerds
- Inner DAG as producer.
BookSearchStreamProducer.executions()runscandidate-discoveryand yields per-node stages. - Selective emission.
select(stage)emits only candidates discovered by the innerdiscover-candidatesstage. - Back-pressure across DAG boundaries.
StreamChannel.driven(..., { capacity: 2 })bounds the outer scatter source. - Archivist domain fixtures. The demo uses real Archivist
CandidateTyperecords and fixture scout batches.
Related Concepts
- Example 34: Intake stream source - Cartographer source-intake gather stream assembly
- Example 20: Streaming execution - per-node AsyncIterable execution
- Streaming Producers - full streaming API: driven, fanIn, resumable, DagStreamProducer