# Streams

> Streams are lazy, single-pass values produced by gen fn . A stream emits values over time and can be consumed with for , .next() , or .iter() .

Website: https://harnlang.com/streams.html

This page documents Harn, which is pre-1.0. Language, standard library, and CLI APIs may change. If the intended version is unclear, clarify before using this page.

---

Streams are lazy, single-pass values produced by `gen fn`. A stream
emits values over time and can be consumed with `for`, `.next()`, or
`.iter()`.

```harn
gen fn numbers(start: int, end: int) -> Stream<int> {
  let n = start
  while n < end {
    emit n
    n = n + 1
  }
}

for n in numbers(1, 4) {
  harness.stdio.log(n)
}
```

`parallel each` can also return a stream instead of waiting for the
whole batch. The streamed form preserves the `max_concurrent` cap but
emits each result as soon as that task completes:

```harn
const results = parallel each [30, 5, 10] with {max_concurrent: 2} { ms ->
  harness.clock.sleep_ms(ms)
  return ms
} as stream

for result in results {
  harness.stdio.log(result) // 5, then 10, then 30
}
```

Use `harness.runtime.parallel_race(items, callable, options?)` when only the
first
successful result matters. It returns the first plain value or
`Result.Ok` payload, cancels remaining work, and throws an aggregate
error when every task throws or returns `Result.Err`.

`gen` is contextual in the `gen fn` declaration form, so existing
identifiers named `gen` remain valid. `emit expr` is only valid inside
`gen fn`. It sends one value to the consumer and then the function
continues when the consumer asks for the next item. Existing `yield`
behavior is unchanged; use `emit` for streams.

`Stream<T>` is distinct from `Generator<T>` in the checker. Regular
functions that already use `yield` keep returning `Generator<T>`.
`gen fn` returns `Stream<T>`.

```harn
gen fn chunks() -> Stream<string> {
  emit "one"
  emit "two"
}

const s: Stream<string> = chunks()
const first = s.next()
harness.stdio.log(first.value)  // one
harness.stdio.log(first.done)   // false
```

Errors thrown inside a stream propagate to the consumer at the point
where the next value is pulled:

```harn
gen fn broken() -> Stream<int> {
  emit 1
  throw "failed"
}

try {
  for n in broken() {
    harness.stdio.log(n)
  }
} catch err {
  harness.stdio.log("caught ${err}")
}
```

Breaking out of a `for` loop stops consuming the stream. Stream
operators such as map/filter/merge/throttle and built-in LLM token
streaming are separate runtime features layered on top of this base
value type.

The `event_log` namespace exposes the active runtime EventLog as stream
values. `event_log.subscribe({topic, from_cursor, kind_prefix?})`
returns a `Stream<dict>` whose events include `{id, cursor, topic, kind,
payload, headers, occurred_at_ms}`. `kind_prefix` filters the stream to
matching event kinds before delivery. Dropping the stream closes the
underlying subscription.

---

## Read next

- [Concurrency](https://harnlang.com/concurrency.md)
- [Runtime context](https://harnlang.com/runtime-context.md)