Module async/stream

Async iteration — the Stream trait: "a value, later, repeatedly" (plans/reference/ASYNC_ITERATION_STREAM.md).

Iterator yields values now; Future yields one value later. A Stream is both: each next(io) returns a FUTURE of Option(Item), and .None means the stream is finished. Before this trait existed, every std API that produced a sequence asynchronously invented its own shape — TcpListener.accept in the caller's own loop, Watcher.next answering Option(FsEvent), Channel.recv plus a separate try_recv — and none of them composed.

{ Stream } :: import("std/async/stream");
{ watch, WatchOptions } :: import("std/fs/watch");

w := watch(Path.new(`./logs`), WatchOptions.defaults(), exn);
names := w.map(e => e.name).take(usize(3));
got := io.await(names.collect(io), io);   // ArrayList(String), 3 entries

The combinators (map, filter, filter_map, take, skip) are LAZY: each wraps the upstream stream and pulls one item per next. The consumers (for_each, collect) drive the stream to .None.

There is no for_await

The plan called for a for_await(stream, io, (x) => body) macro — a loop whose body can break, continue and return. It was written and it deadlocks: an io.await reached only through a MACRO EXPANSION is not counted as a suspension point, so the enclosing io.async body is emitted as a plain closure with a BLOCKING await, and a blocking await inside a spawned task nests the event loop (issues/io-await-inside-a-macro-expansion-is-emitted-as-a-blocking-await.md, plans/backlog/FOR_AWAIT_NEEDS_MACRO_AWARE_ASYNC_TRANSFORM.md). It was removed rather than shipped, because it works from main and hangs in a task — the wrong way round for a server loop.

Until that codegen fix lands, the loop is either for_each (whose await lives in an ordinary function body, so it works everywhere) or the hand-written form, which is what for_await expanded to:

(done : bool) = false;
while(done == false, {
  nx := io.await(stream.next(io), io);
  match(nx, .Some(x) => { … }, .None => { done = true; });
});

Implementing it

next takes self : Self — not inout(self), the way Iterator does — because the future it returns outlives the call, and an inout borrow cannot be held across a suspension. So every stream SOURCE is a reference-semantics type (ref(struct(...)) / ref(enum(...))), and field writes propagate through the handle:

Countdown :: ref(struct(_n : i32));
impl(
  Countdown,
  Stream(
    Item : i32,
    next : (fn(self : Self, io : Io) -> Impl(Future(Option(i32), Io)))(
      io.async((io : Io) =>
        cond(
          (self._n <= i32(0)) => Option(i32).None,
          true => {
            v := self._n;
            self._n = (self._n - i32(1));
            Option(i32).Some(v)
          }
        )
      )
    )
  )
);

A stream whose items can FAIL carries the failure in the item — Item = Result(T, E), the way TcpListener.incoming yields Result(TcpStream, IoError) — so one bad item does not end the stream and no consumer needs an Exception handler. That is why next's future is Future(_, Io) and not Future(_, IoExn): a stream never throws.

Writing a chain

Build the chain OUTSIDE the io.async body that awaits it. An => closure passed to a generic callback parameter INSIDE an async body leaves the enclosing future's result type unresolved (issues/closure-argument-inside-an-io-async-body-loses-the-future-result-type.md), so s.map(f) written inside a task body fails at the task's await. The chain is a value: make it first, then await it wherever you like — including from inside a spawned task.

Stability

unstable — new in v0.2.27; the API may still change until the next release. Two things are most likely to move: the combinator SET (flat_map is deliberately absent — its doubly-derived item type is the shape a var-bound combinator receiver still resolves wrongly), and how the async for loop is eventually spelled (see "There is no for_await" above).