Online edition for BLOGE
0.9.8-RC1· facts verified 2026-09-15 · 中文
Appendix C — Streaming
A graph usually materialises all node outputs before downstream nodes run. Streaming changes that contract: data flows item-by-item through a channel while the upstream node is still producing.
When You Need Streaming
Most BLOGE graphs are "batch-complete": every node finishes, its output is stored, and only then do downstream nodes see that output. This works well when datasets are bounded and fit comfortably in memory.
Streaming is for the cases where that model breaks:
- The upstream produces data incrementally — audio chunks, database cursor pages, live sensor readings — and you want to process items as they arrive.
- The dataset is large enough that materialising it all at once is wasteful — you would rather pipeline through a bounded buffer.
- Latency matters more than throughput — showing partial results to a user while the rest are still being computed.
If none of these apply, regular nodes and foreach are simpler and easier to
test. If the hard part of your workflow is not data volume but lifecycle —
multiple external turns, named states, or signal ownership — continue with
Chapter 14 through
Chapter 16 instead of reaching
for streaming primitives.
The Three Streaming Primitives
BLOGE offers three constructs that produce streaming output. Each serves a different pattern.
1. stream node — a streaming source or transform
A stream node declares an operator whose output is emitted item by item
through an internal NodeChannel rather than returned as a single value.
stream node audioCapture : AudioCapture {
buffer = 64
}
The operator pushes items into the channel at its own pace. Downstream nodes can consume the channel live or wait for it to close and receive the materialised list.
Use when: a single operator produces an unbounded or long-running sequence of items (audio capture, event stream, paginated API).
2. stream foreach — iterate a collection with streaming output
stream foreach is the streaming variant of foreach. It iterates over a
collection and emits each item's result downstream as soon as that item
completes, rather than waiting for every item to finish.
stream foreach processOrders : item in loadOrders.output.orders {
buffer = 16
node processItem : OrderProcessor {
input {
order = item
}
}
}
Use when: you have a bounded collection but want downstream nodes to start consuming results before the last item is processed.
3. stream loop — poll with streaming output
stream loop is the streaming variant of loop. Each iteration's output is
emitted downstream immediately, and the loop continues until until is
satisfied or max_iterations is reached.
stream loop checkStatus {
max_iterations = 100
delay = 2s
depends_on = [initMonitor]
buffer = 8
node pollStatus : StatusPoller {
input {
serviceId = initMonitor.output.serviceId
iteration = loopIteration
}
}
until pollStatus.output.status == "ready"
}
Use when: you are polling or producing items over time and want to forward each result immediately rather than collecting all iterations first.
.stream vs .output — the Two Edge Types
When a downstream node references a streaming node's result, the edge type determines when and how data arrives.
| Syntax | Edge Type | Behavior |
|---|---|---|
upstream.stream | StreamEdge | Downstream receives items live, one by one, as the channel produces them. The downstream operator must be able to process incremental input. |
upstream.output | DirectEdge | Downstream waits until the stream closes, then receives the complete materialised List<T>. Use this when the downstream operator needs the whole dataset. |
Choosing the right edge
Ask: does the downstream operator need all the data before it can start?
- Yes → use
.output. Example: a report generator that needs every processed order to compute totals. - No → use
.stream. Example: a speech-to-text node that transcribes audio chunks as they arrive.
The voice pipeline example shows both edges in a single graph:
/// Receives audio chunks live via StreamEdge
stream node speechToText : SpeechToText {
input {
audio = audioCapture.stream // StreamEdge — live forwarding
}
buffer = 16
}
/// Waits for the full transcript via DirectEdge
node textAnalysis : TextAnalyzer {
depends_on = [speechToText]
input {
transcript = speechToText.output // DirectEdge — materialised List<T>
}
}
Buffer Sizing
Every streaming construct has a buffer parameter that sets the capacity of
the internal NodeChannel ring buffer.
stream node audioCapture : AudioCapture {
buffer = 64 // ring buffer holds up to 64 items
}
Default: 16 items.
How to think about buffer size
The buffer decouples the producer's rate from the consumer's rate. When the buffer is full, the producer blocks (back-pressure) until the consumer catches up.
| Scenario | Guidance |
|---|---|
| Producer is much faster than consumer | Increase the buffer to absorb bursts; back-pressure will still kick in when the gap is too large. |
| Producer and consumer run at similar speeds | Default of 16 is usually fine. |
| Memory is constrained or items are large | Reduce the buffer to limit memory usage; accept more frequent back-pressure pauses. |
| Latency-sensitive pipeline | Keep the buffer small so items are not queued longer than necessary. |
There is no formula — start with the default, measure, and adjust. The engine logs back-pressure events when observability listeners are active, which makes tuning straightforward.
Back-pressure, Cancellation, and Durability Boundary
| Event | Contract | What to test |
|---|---|---|
| Buffer becomes full | Producer waits; items must not be silently dropped | Slow the consumer and observe bounded memory plus eventual progress |
| Downstream cancels | Cancellation must propagate upstream and close the channel | Assert producer termination and no post-cancel emissions |
| Producer fails | Channel closes with failure; materialized .output must not look complete | Assert downstream failure and partial-output policy |
| Process crashes | In-memory channel contents are not a durable checkpoint | Restart and prove what is replayed from source or store |
Streaming and durable execution solve different problems. A bounded channel controls live flow inside a running process; it does not make in-flight items survive a crash. If replay is required, name a durable source offset or checkpoint and define duplicate handling. Do not claim exactly-once delivery from buffer size alone.
Putting It Together — Real Examples
The following files in the repository demonstrate the three streaming primitives in realistic scenarios.
| File | Primitive | What it shows |
|---|---|---|
streaming-batch.bloge | stream foreach | Processes customer orders with streaming output; downstream report uses .output (DirectEdge) to materialise all results. |
streaming-status-monitor.bloge | stream loop | Polls a service at 2-second intervals, streaming each status update downstream; loop exits early when the service reports "ready". |
voice-pipeline.bloge | stream node | Audio capture → speech-to-text → text analysis. Shows both .stream (live) and .output (materialised) edges in the same graph. |
Open these files, read the doc comments, and trace how data flows from the
streaming source through the buffer to downstream consumers. Modify buffer
values and observe how back-pressure behavior changes.
Quick Reference
Common Mistakes
| Mistake | Why It Happens | Fix |
|---|---|---|
Using .stream when the operator needs all data | Habit of choosing "the fast path" | If the operator cannot produce output until it has seen every item, use .output. |
| Setting buffer too large "just in case" | Trying to avoid back-pressure entirely | A huge buffer delays back-pressure signals and wastes memory. Start with the default. |
Forgetting that stream foreach still collects for .output | Assuming streaming means nothing is materialised | Items are streamed and collected; .output gives you the full list after the stream closes. |
| Not testing the back-pressure path | Only testing with small datasets where the buffer never fills | Use MockOperator.delaying(...) in tests to simulate a slow consumer and verify the graph behaves correctly under back-pressure. |
Relationship to the Main Chapters
Streaming is not covered in a single chapter because it builds on several concepts:
- Nodes and dependencies — Chapter 2 and Chapter 3
- Batch and iteration — Chapter 11
introduces
foreachandloop; streaming variants add the channel. - Testing — Chapter 18 shows how to assert on streaming graph results.
- Observability — Chapter 19 covers listeners that log back-pressure events.
This appendix collects the streaming mental model in one place so you can reference it whenever a graph needs incremental data flow.